consumer消费消息后,如何准确回包

简介: consumer消费消息后,如何准确回包

consumer消费消息后,如何准确回包
(1)producer在发送消息的时候,会给每条消息生成唯一的标识符,同时还带上了producer的clientId。当consumer收到并消费消息后,从消息中取出消息的标识符correlationId和producer的标识符clientId,放入响应消息,用来确定此响应消息是哪条请求消息的回包,以及此响应消息应该发给哪个producer。同时响应消息中设置了消息的类型以及响应消息的topic,然后consumer将消息发给broker,如下图所示。

(2)broker收到响应消息后,需要将消息发回给指定的producer。Broker如何知道发回给哪个producer?因为消息中包含了producer的标识符clientId,在ProducerManager中,维护了标识符和channel信息的对应关系,通过这个对应关系,就能把回包发给对应的producer。

响应消息发送和一般的消息发送流程区别在于,响应消息不需要producer拉取,而是由broker直接推给producer。同时选择broker的策略也有变化:请求消息从哪个broker发过来,响应消息也发到对应的broker上。

Producer收到响应消息后,根据消息中的唯一标识符,从RequestResponseFuture的map中找到对应的RequestResponseFuture结构,设置响应消息,同时计数器减一,解除等待状态,使请求方收到响应消息。

目录
相关文章
|
消息中间件 存储 缓存
RabbitMq如何防止消息被重复消费
RabbitMq如何防止消息被重复消费
1359 0
|
消息中间件 数据库
RabbitMQ消息的重复消费问题如何解决的
RabbitMQ消息的重复消费问题如何解决的
1681 0
|
7月前
|
消息中间件 Shell 数据处理
rocket mq 查看消费进度,消息堆积,清除堆积数据命令
该内容是关于RocketMQ的消费进度管理和堆积数据处理的指导。首先,需进入RocketMQ的bin目录,然后使用`mqadmin consumerProgress`命令查看消费者或生产者的消费进度。`broker offset`和`consumer offset`的差值表示未消费消息。通过`resetOffsetByTime`命令可重置消费位点来清除堆积数据,未消费消息默认3天后会被丢弃。此外,`CONSUME_FROM WHERE`枚举类定义了消费起点选项,包括从最后、最开始或指定时间点消费。
1825 3
|
负载均衡 网络性能优化
EMQ如何保证消息不重复消费?
EMQ(Erlang MQTT Broker)通过以下机制来保证消息不重复消费
795 2
|
消息中间件 NoSQL Redis
消息重复消费的问题
消息重复消费的问题
|
消息中间件 NoSQL Kafka
如何保证消息不被重复消费?或者说,如何保证消息消费的幂等性?
为了提高应用程序的性能和可扩展性,很多应用程序开始采用消息队列(MQ)来处理消息。 MQ 可以将消息异步地发送到目的地,从而实现解耦、异步处理和流量控制等功能。 但是,MQ 也带来了一些问题,如消息重复消费和消息消费的幂等性问题。 本文将介绍 MQ 如何保证消息不被重复消费,并讨论如何保证消息消费的幂等性。
|
消息中间件 存储 安全
RocketMQ-消息消费模式 顺序消费
RocketMQ-消息消费模式 顺序消费
226 0
|
消息中间件 Kafka 测试技术
MQ 学习日志(七) 保证消息消费的顺序性
保证消息消费的顺序性
190 0
|
消息中间件
RabbitMQ消息的重复消费问题如何解决的?
RabbitMQ消息的重复消费问题是在分布式系统中常见的一个挑战,解决该问题可以采取以下几种策略:
795 0
|
消息中间件 缓存 数据库
RabbitMQ消息的重复消费问题如何解决的 ?
在RabbitMQ中,重复消费是一个常见的问题,但可以通过以下几种方式来解决
1494 0