Pulsar搭建本地单机环境及测试消息发送与接收

素颜马尾好姑娘i 2021-12-18 23:21 1372阅读 0赞

> ### Pulsar搭建本地单机环境及测试消息发送与接收 ###

**（1）集群环境搭建 **

单机部署官方说明文档：[http://pulsar.apache.org/docs/en/standalone/][http_pulsar.apache.org_docs_en_standalone]

**Pulsar-集群环境搭建与安装部署教程：参考：**[https://www.jianshu.com/p/728d07918f49][https_www.jianshu.com_p_728d07918f49]（本地单机不建议，没耐心做完）

**（2）docker部署pulsar需要环境(无需任何配置比方法（1）简易)**

1.首先部署docker环境：前面有博客讲解详细教程：[https://blog.csdn.net/caox\_nazi/article/details/77451352][https_blog.csdn.net_caox_nazi_article_details_77451352] （centos 配置 docker）

2.**docker pull : **从镜像仓库中拉取或者更新指定镜像

docker pull  apachepulsar/pulsar-standalone (等待时间有点长)

docker 运行镜像

docker run -it -p 28000:80 -p 28080:8080 -p 26650:6650 apachepulsar/pulsar-standalone

**【说明】：**

它会去本地建立一个标准的pulsar server，其中各个端口的意义分别是：

> 80: the port for pulsar dashboard  
> 8080: the http service url for pulsar service  
> 6650: the binary protocol service url for pulsar service

我这边映射到了28000,28080，26650三个端口。

4.其中遇到docker启动失败的问题，解决方案：**[docker启动失败解决][docker]**

**（3）Pulsar-client 消息生产和消费者**

**1.pom.xml 客户端依赖**

<pulsar.version>2.3.2</pulsar.version>
    
    <dependency>
    		<groupId>org.apache.pulsar</groupId>
    		<artifactId>pulsar-client</artifactId>
    		<version>${pulsar.version}</version>		
    </dependency>

**2. 客户端client构建**

package com.caox.pulsar;
    
    import org.apache.pulsar.client.api.PulsarClient;
    import org.apache.pulsar.client.api.PulsarClientException;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/6/13 19:42
     */
    public class Client {
        private PulsarClient client;
    
        public Client() throws PulsarClientException {
            client = PulsarClient.builder()
                    .serviceUrl("pulsar://192.168.80.150:26650")
                    .build();
        }
        public void Close() throws PulsarClientException {
            client.close();
        }
        public PulsarClient getPulsarClient(){
            return client;
        }
    }

**3.1消息生产者MessageProducer.java**  **(发布制定主题)**

package com.caox.pulsar;
    
    import org.apache.pulsar.client.api.Producer;
    import org.apache.pulsar.client.api.PulsarClientException;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/6/13 19:53
     */
    public class MessageProducer {
        private Client client;
        private Producer<byte[]> producer;
    
        public MessageProducer(String topic) throws PulsarClientException {
            client = new Client();
            producer = createProducer(topic);
        }
    
        private Producer<byte[]> createProducer(String topic) throws PulsarClientException {
            return client.getPulsarClient().newProducer()
                    .topic(topic)
                    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                    .sendTimeout(10, TimeUnit.SECONDS)
                    .blockIfQueueFull(true)
                    .create();
        }
    
        public void sendMessage(String message) {
            producer.sendAsync(message.getBytes()).thenAccept(msgId -> {
                System.out.printf("Message with ID %s successfully sent", msgId);
            });
    
        }
        public void sendOnce(String message) {
            /**
             * 发送一次就关闭
             */
            try {
                producer.send(message.getBytes());
                System.out.printf("Message with content %s successfully sent", message);
                producer.close();
                client.Close();
            } catch (PulsarClientException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    
        // todo add exceptionally().
        public void close(Producer<byte[]> producer){
            producer.closeAsync()
                    .thenRun(() -> System.out.println("Producer closed"));
        }
    
        public static void main(String[] args) throws PulsarClientException {
            MessageProducer producer = new MessageProducer("topic1");
    //        producer.sendMessage("Hello World ,lalla");
            producer.sendOnce("Hello World ,lizhenwei");
    
        }
    }

**3.2消息** **发布制定主题 另一种方式：**

package com.caox.pulsar.async;
    
    import lombok.extern.slf4j.Slf4j;
    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;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/6/17 16:53
     * 异步
     */
    @Slf4j
    public class ProducerDemm {
        private static final String SERVER_URL = "pulsar://192.168.80.150:26650";
    
        public static void main(String[] args) throws Exception {
           // 构造Pulsar Client
            PulsarClient client = PulsarClient.builder()
                    .serviceUrl(SERVER_URL)
                    .build();
    
            // 构造生产者
            Producer<String> producer = client.newProducer(Schema.STRING)
                    .producerName("my-producer")
                    .topic("persistent://public/default/my-topic")
                    .batchingMaxMessages(1024)
                    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                    .enableBatching(true)
                    .blockIfQueueFull(true)
                    .maxPendingMessages(512)
                    .sendTimeout(10, TimeUnit.SECONDS)
                    .blockIfQueueFull(true)
                    .create();
    
            // 同步发送消息
            MessageId messageId = producer.send("Hello World");
            log.info("message id is {}", messageId);
            CompletableFuture<MessageId> asyncMessageId = producer.sendAsync("This is a async message");
            // 阻塞线程，直到返回结果
            log.info("async message id is {}",asyncMessageId.get());
    
            // 配置发送的消息元信息，同步发送
    //        producer.newMessage()
    //                .key("my-async-message-key")
    //                .value("my-async-message")
    //                .property("my-async-key", "my-async-value")
    //                .property("my-async-other-key", "my-async-other-value")
    //                .sendAsync();
    
            // 关闭producer的方式有两种：同步和异步
             producer.close();
    
            // 关闭licent的方式有两种，同步和异步
            client.close();
    
        }
    
    }

**4.1消息消费者MessageConsumer.java** **（单主题订阅）**

package com.caox.pulsar;
    
    import org.apache.pulsar.client.api.Consumer;
    import org.apache.pulsar.client.api.Message;
    import org.apache.pulsar.client.api.PulsarClientException;
    import org.apache.pulsar.client.api.SubscriptionType;
    
    import java.util.concurrent.CompletableFuture;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/6/13 19:47
     */
    public class MessageConsumer {
        private Client client;
        private Consumer consumer;
    
        public MessageConsumer(String topic, String subscription) throws PulsarClientException {
            client = new Client();
            consumer = createConsumer(topic, subscription);
        }
    
        private Consumer createConsumer(String topic, String subscription) throws PulsarClientException {
    
            return client.getPulsarClient().newConsumer().topic(topic).subscriptionName(subscription)
                    .ackTimeout(10, TimeUnit.SECONDS).subscriptionType(SubscriptionType.Exclusive).subscribe();
        }
    
        public String getMessage() throws ExecutionException, InterruptedException, PulsarClientException {
            /***
             * 获取一次，就关闭会话
             */
            // Wait for a message
            System.out.printf("Start pulsar");
            CompletableFuture<Message> msg = consumer.receiveAsync();
    
            // System.out.printf("Message received: %s", new String(msg.get().getData()));
            String result = "topic is: " + msg.get().getTopicName() + ",data is: " + new String(msg.get().getData());
    
            // Acknowledge the message so that it can be deleted by the message broker
            consumer.acknowledge(msg.get());
            consumer.close();
            client.Close();
            return result;
        }
    
        public void receiveMessage() throws ExecutionException, InterruptedException, PulsarClientException {
            /***
             * 用来异步获取，保持回话
             */
            do {
                // Wait for a message
                CompletableFuture<Message> msg = consumer.receiveAsync();
    
                System.out.printf("Message received: %s", new String(msg.get().getData()));
    
                // Acknowledge the message so that it can be deleted by the message broker
                consumer.acknowledge(msg.get());
            } while (true);
        }
    
        public static void main(String[] args) throws PulsarClientException, ExecutionException, InterruptedException {
            MessageConsumer consumer = new MessageConsumer("topic1", "my-sub");
            consumer.receiveMessage();
    //		String reString = consumer.getMessage();
    //		System.err.println(reString);
            // consumer.client.Close();
    
        }
    }

**4.2 消息 单主题订阅 （另一种方式）**

package com.caox.pulsar.async;
    
    import lombok.extern.slf4j.Slf4j;
    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.CompletableFuture;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/6/17 17:08
     */
    @Slf4j
    public class OnceConsumer {
        private static final String SERVER_URL = "pulsar://192.168.80.150:26650";
    
        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("my-consumer")
                    .topic("my-topic")
                    .subscriptionName("my-subscription")
                    .ackTimeout(10, TimeUnit.SECONDS)
                    .maxTotalReceiverQueueSizeAcrossPartitions(10)
                    .subscriptionType(SubscriptionType.Exclusive)
                    .subscribe();
    //        do {
    //            // 接收消息有两种方式：异步和同步
                 CompletableFuture<Message<String>> message = consumer.receiveAsync();
    //            Message message = consumer.receive();
    //            log.info("get message from pulsar cluster,{}", message);
    //        } while (true);
    
            do {
                // Wait for a message
                CompletableFuture<Message> msg = consumer.receiveAsync();
    
                System.out.printf("Message received: %s", new String(msg.get().getData()));
    
                // Acknowledge the message so that it can be deleted by the message broker
                consumer.acknowledge(msg.get());
            } while (true);
        }
    }

**5. 消费多主题订阅**（**多主题**）

package com.caox.pulsar.async;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.pulsar.client.api.Consumer;
    import org.apache.pulsar.client.api.ConsumerBuilder;
    import org.apache.pulsar.client.api.Message;
    import org.apache.pulsar.client.api.PulsarClient;
    
    import java.util.Arrays;
    import java.util.List;
    import java.util.concurrent.CompletableFuture;
    import java.util.regex.Pattern;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/6/17 17:19
     */
    @Slf4j
    public class MultiConsumer {
    
        private static final String SERVER_URL = "pulsar://192.168.80.150:26650";
        private static final String DEFAULT_NS_TOPICS = "persistent://public/default/.*";
        private static final String DEFATULT_NS_REG_TOPICS= "persistent://public/default/my1.*";
    
        @SuppressWarnings("all")
        public static void main(String[] args) throws Exception {
            PulsarClient client = PulsarClient.builder()
                    .serviceUrl(SERVER_URL)
    //                .enableTcpNoDelay(true)
                    .build();
            ConsumerBuilder consumerBuilder = client.newConsumer()
                    .subscriptionName("multi-sub");
    
            // 方式一： 订阅namespace下所有的topic
    //        Pattern allTopicsInNamespace = Pattern.compile(DEFAULT_NS_TOPICS);
    //        Consumer allTopicsConsumer =  consumerBuilder.topicsPattern(allTopicsInNamespace).subscribe();
    
            // 方式二： 订阅namespace下满足正则匹配的topic
            Pattern someTopicsInNamespace = Pattern.compile(DEFATULT_NS_REG_TOPICS);
    
            Consumer allTopicsConsumer = consumerBuilder
                    .topicsPattern(someTopicsInNamespace)
                    .subscribe();
    
            // 方式三：自定义topic用方式一和二，则.topic(必须空或者null)topics
            List<String> topics = Arrays.asList(
                    "topic-1",
                    "topic-2",
                    "topic-3",
                    "my-topic"
            );
    
    //        Consumer multiTopicConsumer = consumerBuilder
    //                .topics(topics)
    //                .subscribe();
    
            do {
                // 接收消息有两种方式：异步和同步
                CompletableFuture<Message<String>> message = allTopicsConsumer.receiveAsync();
                log.info("Message received: {}", new String(message.get().getData()));
                // Acknowledge the message so that it can be deleted by the message broker
                allTopicsConsumer.acknowledge(message.get());
            } while (true);
    
        }
    }

**其中 （1）消息发送与接收 client关闭 均存在两种方式同步和异步；**

**        （2）多主题订阅消费包含三种topic主题：**

**                  1. 订阅namespace下所有的topic；**

**                  2. 订阅namespace下满足正则匹配的topic;**

**                  3. 自定义topic用方式一和二，则.topic(必须空或者null)topics**

[http_pulsar.apache.org_docs_en_standalone]: http://pulsar.apache.org/docs/en/standalone/
[https_www.jianshu.com_p_728d07918f49]: https://www.jianshu.com/p/728d07918f49
[https_blog.csdn.net_caox_nazi_article_details_77451352]: https://blog.csdn.net/caox_nazi/article/details/77451352
[docker]: https://www.cnblogs.com/SimonHu1993/p/9199566.html