Messages

• Pulsar 中的基本单位
• 包含以下部分:
  • Value:消息携带的数据
  • Key:消息 Key，可以用于主题合并（同个 Key 保留最新消息）
  • Properties:用户自定义配置
  • Producer name:生产者名称，没有声明使用默认的
  • Topic name:消息将要发送到的 topic name
  • Schema version:Schema 版本
  • Sequence ID:针对于 topic 级别，每条消息的序号，可以用于消息去重，如果 brokerDeduplicationEnabled 设置为 true，该值在每个非分区 topic 中是唯一的
  • Message ID:bookies 中分配的消息 ID，表示每个消息在 ledger 中的位置，并且在单个 Pulsar 集群中唯一
  • Publish time:消息发布时间
  • Event time:通过应用额外和消息绑定的时间，没有设置默认为 0
• 消息默认大小为 5 MB，可以通过以下配置修改:
  • broker.conf 文件:maxMessageSize=5242880
  • bookkeeper.conf 文件:nettyMaxFrameSizeBytes=5253120

生产者

发送模式

• 同步
• 异步:消息先放进阻塞队列然后立即返回，客户端在后台异步发送，队列满了生产者会阻塞或者失败

访问模式

• Shared:允许多个生产者同时发布消息给主题，默认
• Exclusive:只允许一个生产者发送消息给主题，其他生产者连接会报错
• WaitForExclusive:同上，但是不报错只是阻塞，等连接的生产者下线后，会有一个被阻塞的生产者连接，其他的继续阻塞

压缩

• LZ4
• ZLIB
• ZSTD
• SNAPPY

批

• 批大小由消息最大数量和最大发布延迟共同决定
• backlog size 代表所有批的数量而不是所有消息的数量
• 批次作为单个单元而不是单个消息进行跟踪和存储，消费者将一个批次拆分为单独的消息，但是即使配置了批处理，如果配置了 deliverAt 或 deliverAfter 参数，消息还是会单独发送
• 通常来讲如果一批次的消息中有一部分投递失败，则整个批次会被重新投递，为了解决该问题，Pulsar 2.6.0 之后引入了 batch index acknowledgement，会过滤已经消费的消息，直到整个批次的消息全部确认，删除该批次，通过修改 acknowledgmentAtBatchIndexLevelEnabled=false 为true来启用该机制，默认关闭，如果启用了该机制，会增大内存负载

Chunking

• 生产者对大消息进行拆分，消费者聚合
• 局限:
• 仅适用于持久化主题
• 仅适用于独享和灾备订阅模式
• 不能和批处理同时启用
• 消费端会对 chunk 进行缓存，收到所有 chunk 才进行聚合
• 通过配置 maxPendingChunkedMessage 参数来限制消费者同时维护的分块消息的最大数量
• 启用:
• 禁用批机制，enableBatching 设置为 false
• 默认关闭，设置 chunkingEnabled 为 true 开启（创建生产者时）
• 如果消费者在 1 分钟没有收到所有的 chunks ，会将不完整的 chunk 过期，通过设置 expireTimeOfIncompleteChunkedMessage 参数可以调用（ ConsumerBuilder 构建时可以调用 ConsumerBuilder.expireTimeOfIncompleteChunkedMessage​(long duration, java.util.concurrent.TimeUnit unit) 进行设置）

消费者

消费端维护一个队列接收消息，大小通过 receiverQueueSize 来配置，默认为 1000，每次调用 consumer.receive() 方法从队列中取出一条数据

接收模式

• 同步
• 异步

监听器

客户端为消费者实现监听器，比如 Java 客户端提供了 MessageListener 接口，每次收到新消息的时候都会调用 received 方法

消息确认

• 单条确认: consumer.acknowledge(msg)
• 累计确认: consumer.acknowledgeCumulative(msg)，不能用于共享订阅模式，因为会涉及多个消费者，所以只能单条确认

取消确认

• 对失败的消息调用 negativeAcknowledge 方法取消确认，要求 Broker 重新投递

确认超时

• ackTimeout:确认超时时间
• ackTimeoutTickTime:检查间隔，默认 1 秒

重试主题

• 消息处理失败，消费者可以将消息发送给重试主题，重新消费
• 默认情况禁用自动重试，通过将 enableRetry 设置为 true
• 当达到 maxRedeliverCount 时，未消费的消息将会被移到死信主题
• 通过下列方式使用:

Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Shared)
        .enableRetry(true)
        .deadLetterPolicy(DeadLetterPolicy.builder()
                .maxRedeliverCount(maxRedeliveryCount)
                .retryLetterTopic("my-retry-letter-topic-name")
                .build())
        .subscribe();

死信主题

• 存储消费失败的消息
• 使用方式如下:

Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .subscriptionType(SubscriptionType.Shared)
                .deadLetterPolicy(DeadLetterPolicy.builder()
                      .maxRedeliverCount(maxRedeliveryCount)
                      .deadLetterTopic("my-dead-letter-topic-name")
                      .initialSubscriptionName("init-sub")
                      .build())
                .subscribe();

主题

• Pulsar支持持久化主题和非持久化主题，前缀分别为persistent和non-persistent
• Pulsar提供默认的命名空间default，默认租户public

订阅

订阅类型

• Exclusive:独享，一个订阅组只绑定一个消费者
• FailOver:灾备，一个订阅组绑定多个消费者，Pulsar从中选择一个作为主消费者，将所有的消息发送给该消费者，该消费者因故下线，重新选择主消费者
• Shared:共享，一个订阅组绑定多个消费者，轮询方式分发消息
• Key_Shared:键共享，相同key发送到同一个消费者（该模式生产者需禁用消息批次机制或使用 BatcherBuilder.KEY_BASED，可以将相同键的消息放到一个批次中）

订阅模式

• 持久:默认情况下，没有任何持久订阅的消息会被标记为已删除，如果为该主题创建了持久订阅，只有确认的消息才会被标记为已删除
• 非持久
• 使用方式:

Consumer<byte[]> consumer = pulsarClient.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-sub")
        .subscriptionMode(SubscriptionMode.Durable)
        .subscribe();

多主题订阅

• 正则表达式（所以主题必须在同一个命名空间）
• 主题列表

分区主题

实际是多个内部主题，每个分区绑定一个 Broker

路由模式

决定每条消息被发送到哪个内部主题

• RoundRobinPartition:如果消息指定key，按照key的Hash值将该消息分配给对应的内部主题，如果没有指定key，会将一个时间段内没有key的消息发送给同一个内部主题，每隔10ms以轮询方式切换到下一个内部主题
• SinglePartition:如果消息指定key，按照key的Hash值将该消息分配给对应的内部主题，如果没有指定key，随机选择一个内部主题
• CustomPartition:自定义路由器

Hashing scheme

• 枚举类，表示选择消息分区时可以使用的散列函数的集合
• 有两种类型:
  • JavaStringHash:默认
  • Murmur3_32Hash:当生产者来自不同多种语言的客户端时，推荐用该类型

非持久化主题

• 消息不会持久化到磁盘，只存在于内存中的主题
• 如果 Broker 掉线或者断开该主题的订阅，消息将会丢失
• 非持久化主题中，Broker 会立即将消息投递给所有的订阅者而不会持久化到 BookKeeper，这样在某些情况下可以使得消息投递速度比持久化主题快，但是会丢失 Pulsar 的一些核心优势
• 默认情况下启用

系统主题

• Pulsar 内部使用的预先定义的主题，可以是非持久化或者持久化主题
• 系统主题用于实现某些功能并且消除对第三方组件的依赖，比如事务、心跳检测、主题等级策略和资源组服务等，系统主题使得这些功能的实现变得简单、可靠和灵活
• 不能创建任何系统主题
• 通过该链接查看有哪些可用的系统主题:https://pulsar.apache.org/docs/concepts-messaging#system-topic
• 默认情况下，系统主题被禁用，可以通过改变 conf/broker.conf 或 conf/standalone.conf 文件中的以下配置启用

systemTopicEnabled=true
topicLevelPoliciesEnabled=true

消息去重

生产者幂等

• 确保每条消息只生成一次
• 缺点事将重复消息的删除工作放在应用层面处理
• 无需修改客户端代码，只需要修改某些配置

重复数据删除和单次有效语义

参考该链接:https://www.splunk.com/en_us/blog/it/exactly-once-is-not-exactly-the-same.html

原文地址：http://www.cnblogs.com/jordan95225/p/16809864.html