原创

Pulsar正则匹配订阅topic

温馨提示：
本文最后更新于 2020年07月29日，已超过 1,577 天没有更新。若文章内的图片失效（无法正常加载），请留言反馈或直接联系我。

Pulsar在很多情况下提供了比Kafka 更快的吞吐量和更低的延迟，并为开发人员提供了一组兼容的API，让他们可以很轻松地从Kafka 切换到Pulsar。 Pulsar 的最大优点在于它提供了比Apache Kafka 更简单明了、更健壮的一系列操作功能，特别在解决可观察性、地域复制和多租户方面的问题。

Pulsar 是一个用于服务器到服务器的消息系统，具有多租户、高性能等优势。 Pulsar 最初由 Yahoo 开发，目前由 Apache 软件基金会管理。
Pulsar 的关键特性如下：

  1. Pulsar 的单个实例原生支持多个集群，可跨机房在集群间无缝地完成消息复制。
  2. 极低的发布延迟和端到端延迟。
  3. 可无缝扩展到超过一百万个 topic。
  4. 简单的客户端 API，支持 Java、Go、Python 和 C++。
  5. 支持多种 topic 订阅模式（独占订阅、共享订阅、故障转移订阅）。
  6. 通过 Apache BookKeeper 提供的持久化消息存储机制保证消息传递 。
    • 由轻量级的 serverless 计算框架 Pulsar Functions 实现流原生的数据处理。
      基于 Pulsar Functions 的 serverless connector 框架 Pulsar IO 使得数据更易移入、移出 Apache Pulsar。

分层式存储可在数据陈旧时，将数据从热存储卸载到冷/长期存储（如S3、GCS）中。

  1. 消息模型
    Pulsar 提供了统一的消息模型和 API。streaming 模式——独占和故障切换订阅方式; queuing 模式——共享订阅的方式。
    Kafka 主要集中在 streaming 模式，对单个 partition 是独占消费，没有共享（Queue）的消费模式；

  2. 架构
    Pulsar Broker是无状态的，与存储相互分离 可以轻松添加和删除节点，而无需重新平衡整个集群
    Kafka的数据直接存储在Broker上(有状态的) 任何容量扩展都需要重新平衡分区，同时还需要将被平衡的分区重新拷贝到新添加的Broker上

  3. Ack
    Pulsar使用专门的 Cursor 管理。累积确认(cumulative acknowledgment)和 Kafka 效果一样；提供单条确认(individual ack)。
    Kafka使用偏移 Offset

  4. Retention
    Pulsar 消息只有被所有订阅消费后才会删除，不会丢失数据。也允许设置保留期，保留被消费的数据。支持 TTL
    Kafka根据设置的保留期来删除消息。有可能消息没被消费，过期后被删除。 不支持 TTL。

参考链接：
https://pulsar.apache.org/docs/zh-CN/concepts-overview/
http://lxkaka.wang/2019/03/25/pulsar/

Pulsar正则匹配订阅topic

1.Maven依赖

    <properties>
        <junit.version>4.12</junit.version>
        <pulsar.version>2.5.2</pulsar.version>
        <slf4j-log4j12.version>1.7.30</slf4j-log4j12.version>
        <log4j.version>1.2.17</log4j.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.pulsar</groupId>
            <artifactId>pulsar-client</artifactId>
            <version>${pulsar.version}</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>${junit.version}</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j-log4j12.version}</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>${log4j.version}</version>
        </dependency>
    </dependencies>

2.创建一个client

package com.lzhpo.listentopic;

import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;

/**
 * Pulsar客户端
 * 
 * @author lzhpo
 */
public class Client {

    private PulsarClient client;

    public Client() throws PulsarClientException {
        client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
//                .authentication(AuthenticationFactory.token("123"))
                .build();
    }
    public void Close() throws PulsarClientException {
        client.close();
    }
    public PulsarClient getPulsarClient(){
        return client;
    }
}

3.创建一个消费者进行测试

package com.lzhpo.listentopic;

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;
import java.util.regex.Pattern;

/**
 * Pulsar消费者
 *
 * @author lzhpo
 */
public class PulsarConsumerApp {

    private Client client;
    private Consumer<byte[]> consumer;

    /**
     * 构造器初始化Client
     *
     * @throws PulsarClientException
     */
    public PulsarConsumerApp() throws PulsarClientException {
        client = new Client();
    }

    /**
     * 创建消费者
     *
     * @param topic
     * @param subscription
     * @return
     * @throws PulsarClientException
     */
    private void createConsumerByTopic(String topic, String subscription) throws PulsarClientException {
        consumer = client.getPulsarClient().newConsumer()
                .topic(topic)
                .subscriptionName(subscription)
                .ackTimeout(10, TimeUnit.SECONDS)
                .subscriptionType(SubscriptionType.Exclusive)
                .subscribe();
    }

    /**
     * 正则匹配订阅的topic
     *
     * @param topicsPattern
     * @param subscription
     * @throws PulsarClientException
     */
    private void createConsumerByTopicsPattern(String topicsPattern, String subscription) throws PulsarClientException {
        consumer = client.getPulsarClient().newConsumer()
                .topicsPattern(topicsPattern)
                .subscriptionName(subscription)
                .ackTimeout(10, TimeUnit.SECONDS)
                .subscriptionType(SubscriptionType.Exclusive)
                .subscribe();
    }

    /**
     * 1.持续接收消息
     *
     * @throws ExecutionException
     * @throws InterruptedException
     * @throws PulsarClientException
     */
    public void receiveMessageAlways() throws ExecutionException, InterruptedException, PulsarClientException {
        // 用来异步获取，保持回话
        do {
            // 异步接收消息
            CompletableFuture<Message<byte[]>> msg = consumer.receiveAsync();

            System.out.println("Message received from ["+msg.get().getTopicName()+"] topic: " +new String(msg.get().getData()));

            // 确认消息，以便消息代理可以将其删除
            consumer.acknowledge(msg.get());
        } while (true);
    }

    /**
     * 2.获取一次，就关闭会话
     *
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     * @throws PulsarClientException
     */
    public void receiveMessageOnce() throws ExecutionException, InterruptedException, PulsarClientException {
        // 异步接收
        CompletableFuture<Message<byte[]>> msg = consumer.receiveAsync();
        System.out.println("Message received from ["+msg.get().getTopicName()+"] topic: " +new String(msg.get().getData()));

        // 确认消息，以便消息代理可以将其删除
        consumer.acknowledge(msg.get());
        consumer.close();
        client.Close();
    }

    /**
     * @param args
     * @throws PulsarClientException
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void main(String[] args) throws PulsarClientException, ExecutionException, InterruptedException {
        // 持续接收消息
        PulsarConsumerApp consumer = new PulsarConsumerApp();
        consumer.createConsumerByTopicsPattern("persistent://public/default/.*", "test2-sub");
        consumer.receiveMessageAlways();

        // 接收一次消息就关闭
//        PulsarConsumerApp consumer = new PulsarConsumerApp();
//        consumer.createConsumerByTopic(
//                "persistent://public/default/gluon-schema-AvAccountStatusRequest",
//                "test2-sub");
//        consumer.receiveMessageOnce();
    }
}

4.测试结果

当你往pulsar的topic发消息的时候，这边就会显示对应的topic和数据。

pulsar订阅多个topic.png

Pulsar生产者

package com.lzhpo.listentopic;

import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;

import java.util.concurrent.TimeUnit;

/**
 * Pulsar生产者
 * 
 * @author lzhpo
 */
public class PulsarProducerApp {

    private Client client;
    private Producer<byte[]> producer;

    /**
     * 构造器初始化生产者
     *
     * @param topic 主题
     * @throws PulsarClientException
     */
    public PulsarProducerApp(String topic) throws PulsarClientException {
        client = new Client();
        producer = createProducer(topic);
    }

    /**
     * 构建一个生产者
     *
     * @param topic 主题
     * @return
     * @throws PulsarClientException
     */
    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();
    }

    /**
     * 1.异步发送消息
     *
     * @param message
     */
    public void sendMessage(String message) {
        producer.sendAsync(message.getBytes()).thenAccept(msgId -> {
            System.out.printf("Message with ID %s successfully sent", msgId);
        });

    }

    /**
     * 2.发送一次就关闭
     *
     * @param message
     */
    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) {
            e.printStackTrace();
        }
    }

    /**
     * 手动异步关闭
     *
     * @param producer
     */
    public void close(Producer<byte[]> producer){
        producer.closeAsync()
                .thenRun(() -> System.out.println("Producer closed"));
    }

    public static void main(String[] args) throws PulsarClientException {
        PulsarProducerApp producer = new PulsarProducerApp("persistent://public/default/test01-topic");
//        producer.sendMessage("Hello,I'm lewis.");
        producer.sendOnce("Hello,lzhpo");
    }
}
所属分类：Java
热门推荐
相关文章
关于我
会打篮球的程序猿
会打篮球的程序猿
Talk is cheap, show me the code.
本文目录
    近期评论
    标签云
    Linux Java Spring SpringBoot Hadoop Hive CDH RabbitMQ Docker Kafka Zookeeper 面试 数据结构与算法 Eureka SpringCloud Ribbon Feign Hystrix Golang Maven MySQL Redis Ambari OpenStack DataNode ElasticSearch HBase JavaScript Gitlab MapReduce Mybatis Spark 机器学习 Beeline Scala Dubbo KVM IDEA Nginx SEO 资源分享 新闻热点 JS Android Flutter Vue Nodejs Turbine 大学 Windows 碎碎念 Python JVM 设计模式 GraphQL go-micro Flink Pulsar go-chi ClickHouse 分布式 SpringSecurity JPA OAuth2 ChatGPT Storm