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
标签:持久,主题,默认,topic,消息,Messaging,Pulsar From: https://www.cnblogs.com/jordan95225/p/16809864.html