1、Pulsar的单机模式与集群模式安装以及Java客户端消息生产消费的初体验

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.SubscriptionType;
import java.util.concurrent.TimeUnit;

public class PulsarConsumer {

    private static final String SERVER_URL = "pulsar://192.168.0.61:6650,192.168.0.62:6650,192.168.0.63:6650";

    public static void main(String[] args) throws Exception {
        // 构造Pulsar Client
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(SERVER_URL)
                .enableTcpNoDelay(true)
                .build();

        Consumer consumer = client.newConsumer()
                .consumerName("pulsar-consumer")
                .topic("persistent://public/default/hello-world")
                .subscriptionName("my-first-subscription")
                .ackTimeout(10, TimeUnit.SECONDS)
                .maxTotalReceiverQueueSizeAcrossPartitions(10)
                .subscriptionType(SubscriptionType.Exclusive)
                .subscribe();
        do {
            // 接收消息有两种方式：异步和同步
            // CompletableFuture<Message<String>> message = consumer.receiveAsync();
            Message message = consumer.receive();
            System.out.println(String.format("接收到消息[messageId=%s, messageKey=%s, content=%s, properties=%s]",
                    message.getMessageId().toString(),
                    message.getKey(),
                    new String(message.getData()),
                    message.getProperties().toString()
                    ));
            // 消息确认
            consumer.acknowledge(message);
        } while (true);
    }
}

import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class PulsarProducer {

    private static final String SERVER_URL = "pulsar://192.168.0.61:6650,192.168.0.62:6650,192.168.0.63:6650";

    public static void main(String[] args) throws Exception {
        // 构造Pulsar Client
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(SERVER_URL)
                .enableTcpNoDelay(true)
                .build();
        // 构造生产者
        Producer<String> producer = client.newProducer(Schema.STRING)
                .producerName("pulsar-producer")
                .topic("persistent://public/default/hello-world")
                .batchingMaxMessages(1024)
                .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                .enableBatching(true)
                .blockIfQueueFull(true)
                .maxPendingMessages(512)
                .sendTimeout(10, TimeUnit.SECONDS)
                .blockIfQueueFull(true)
                .create();
        // 同步发送消息
        MessageId messageId = producer.send("你好呀，pulsar!");
        System.out.println("消息ID：" + messageId);

        CompletableFuture<MessageId> asyncMessageId = producer.sendAsync("这是一条异步消息");
        // 阻塞线程，直到返回结果
        System.out.println("异步消息ID = " + asyncMessageId.get());

        // 配置发送的消息元信息，同步发送
        producer.newMessage()
                .key("my-message-key")
                .value("配置消息元数据同步发送消息")
                .property("my-key", "my-value")
                .property("my-other-key", "my-other-value")
                .send();
        producer.newMessage()
                .key("my-async-message-key")
                .value("配置消息元数据异步发送消息")
                .property("my-async-key", "my-async-value")
                .property("my-async-other-key", "my-async-other-value")
                .sendAsync();

        // 关闭producer的方式有两种：同步和异步
        // producer.closeAsync();
        producer.close();

        // 关闭licent的方式有两种，同步和异步
        // client.close();
        client.closeAsync();
    }
}

producer.newMessage()
                .key("my-async-message-key")
                .value("配置消息元数据异步发送消息")
                .property("my-async-key", "my-async-value")
                .property("my-async-other-key", "my-async-other-value")
                .sendAsync();