189 8069 5689

Pulsar消息概念1

Pulsar官方文档 概念和架构-Messaging Concepts中主要内容

成都创新互联公司服务项目包括鲤城网站建设、鲤城网站制作、鲤城网页制作以及鲤城网络营销策划等。多年来,我们专注于互联网行业,利用自身积累的技术优势、行业经验、深度合作伙伴关系等,向广大中小型企业、政府机构等提供互联网行业的解决方案,鲤城网站推广取得了明显的社会效益与经济效益。目前,我们服务的客户以成都为中心已经辐射到鲤城省份的部分城市,未来相信会继续扩大服务区域并继续获得客户的支持与信任!

1 消息组成

组成说明
Value / data payload 消息携带的数据,所有pulsar的消息携带原始bytes,但是消息数据也需要遵循数据shcema
Key 消息可以被Key打标签。这可以对topic压缩之类的事情起作用
Properties 用户定义属性的可选键/值映射
Producer name 生成消息的生产者的名称(生产者自动被赋予默认名称,但您也可以显式地应用自己的名称)
Sequence ID 每一个消息在其主题上都属于一个有序序列。消息的序列ID是它在该序列中的顺序
Publish time 消息发布时的时间戳(由生产者自动应用)
Event time 一个可选的时间戳,应用程序可以附加到消息上,表示某个事件发生的时间,例如消息被处理的时间。如果未显式设置,则消息的事件时间为0。

2 生产者 发送模式

模式说明
同步发送 生产者将在发送每个消息后等待代理的确认。如果未收到确认,则生产者将认为发送操作失败
异步发送 生产者将把消息放入阻塞队列并立即返回。客户端库随后将消息发送到后台的代理。如果队列已满(最大大小可配置,则在调用API时,生产者可能会被阻止或立即失败,具体取决于传递给生产者的参数

2.1 消息压缩支持
LZ4
ZLIB
ZSTD
SNAPPY

2.2 支持批处理
如果启用批处理,则生产者将在单个请求中累积并发送一批消息。批处理大小是由消息的最大数量和最大发布延迟定义的。
3 消费者 接收模式

模式说明
同步接收 同步接收将被阻止,直到有消息可用
异步接收 异步接收将立即返回一个future值,例如Java中的CompletableFuture,该值在新消息可用时完成

3.1 监听
客户端库为用户提供侦听器实现。例如,Java客户机提供了一个MessageListener接口。在这个接口中,只要接收到新消息,就会调用received方法。
3.2 确认
当使用者成功使用消息时,使用者会向代理发送确认请求,以便代理丢弃该消息。否则,它将存储消息。
消息可以逐个确认,也可以累积确认。对于累积确认,消费者只需要确认它收到的最后一条消息。流中直至(包括)所提供消息的所有消息将不会重新传递给该使用者
( 累积确认不能与共享订阅模式一起使用,因为共享模式涉及多个对同一订阅具有访问权限的使用者)
在共享订阅模式下,可以单独确认消息。
3.3 否定确认
当使用者一次未成功使用消息,并且希望再次使用该消息时,使用者可以向代理发送否定的确认,然后代理将重新传递该消息。

消息可以被一个接一个地否定或累积地承认,这取决于消费订阅模式。
在独占订阅模式和故障转移订阅模式中,消费者只会消极地确认他们收到的最后一条消息。

在“共享”和“密钥共享”订阅模式中,您可以分别对消息进行否定性确认

3.4 确认超时
当消息未成功使用,并且您希望触发代理自动重新传递消息时,可以采用未确认消息自动重新传递机制。客户端将在整个确认超时时间范围内跟踪未确认消息,并在指定确认超时时自动向代理发送重新传递未确认消息请求
注意
在确认超时之前使用否定确认。否定确认以更精确的方式控制单个消息的重新传递,并在消息处理时间超过确认超时时避免无效的重新传递。
4 死信主题
在某些消息无法由消费者使用时,会成生新的消息。在这种机制中,无法使用的消息存储在一个单独的主题中,称为死信主题。您可以决定如何处理死信主题中的消息
下面的示例演示如何使用默认的死信主题在Java客户机中启用死信主题

Consumer consumer = pulsarClient.newConsumer(Schema.BYTES)
.topic(topic)
.subscriptionName("my-subscription")
.subscriptionType(SubscriptionType.Shared)
.deadLetterPolicy(DeadLetterPolicy.builder()
.maxRedeliverCount(maxRedeliveryCount)
.build())
.subscribe();
默认的死信主题使用以下格式:
--DLQ
如果要指定死信主题的名称,请使用以下Java客户端示例:
Consumer consumer = pulsarClient.newConsumer(Schema.BYTES)
.topic(topic)
.subscriptionName("my-subscription")
.subscriptionType(SubscriptionType.Shared)
.deadLetterPolicy(DeadLetterPolicy.builder()
.maxRedeliverCount(maxRedeliveryCount)
.deadLetterTopic("your-topic-name")
.build())
.subscribe();
死信主题取决于邮件的重新传递。由于确认超时或否定确认,消息将重新传递。如果要对消息使用否定确认,请确保在确认超时之前对其进行否定确认。
注意
目前,死信主题仅在共享订阅模式下启用


当前名称:Pulsar消息概念1
文章分享:http://jkwzsj.com/article/pjjoji.html

其他资讯