网站友链是什么情况,wordpress主题在线编辑,网站建设包含二级网站,备案信息如何上传的网站上RocketMQ常见问题 文章目录 RocketMQ常见问题一#xff1a;消息幂等问题1#xff1a;什么是消费幂等2#xff1a;消息重复的场景分析2.1#xff1a;发送时消息重复2.2#xff1a;消费时消息重复2.3#xff1a;Rebalance时消息重复 3#xff1a;通用解决方案3.1#xff…RocketMQ常见问题 文章目录 RocketMQ常见问题一消息幂等问题1什么是消费幂等2消息重复的场景分析2.1发送时消息重复2.2消费时消息重复2.3Rebalance时消息重复 3通用解决方案3.1两要素(幂等令牌 唯一性处理)3.2通用解决方案3.3解决方案举例3.4消息幂等的实现 二消息堆积问题1产生原因分析1.1消息拉取1.2消息消费 2消费耗时3消费并发度4单机线程的计算5如何避免消息堆积和消费延迟5.1梳理消息的消费耗时5.2设置消费并发度 三消息清理问题四消息过滤问题1Tag方式过滤-简单2SQL过滤-功能强大3代码举例3.1Tag方式进行过滤例子3.2SQL方式进行过滤例子 五消息重试问题1消息发送的重试机制1.1同步发送失败策略1.2异步发送失败策略1.3消息刷盘失败策略 2消费重试2.1顺序消息的消费重试2.2无序消息的消费重试2.3消息重试次数和间隔2.4重试队列2.5消息重试/不重试配置方式2.5.1重试配置2.5.2不重试配置 2.6死信队列2.6.1死信队列特征2.6.2查询死信消息2.6.3死信队列处理 六如何保证消息不丢失1Producer如何保证消息不丢失2事务消息保证消息不丢失3broker保证消息不丢失4Broker主从同步如何保证不丢失5Consumer如何保证消费不丢失6如果MQ服务全挂了如何保证不丢失 一消息幂等问题
1什么是消费幂等
重复消费的结果与消费一次的结果是相同的并且多次消费并未对业务系统产生任何负面影响那么这个消费过程就是消费幂等的。 幂等若某操作执行多次与执行一次对系统产生的影响是相同的则称该操作是幂等的。 在互联网应用中尤其在网络不稳定的情况下消息很有可能会出现重复发送或重复消费。
如果重复的消息可能会影响业务处理那么就应该对消息做幂等处理。
2消息重复的场景分析
什么情况下可能会出现消息被重复消费呢最常见的有以下三种情况
2.1发送时消息重复
当一条消息已被成功发送到Broker并完成持久化此时出现了网络闪断从而导致Broker对Producer应答失败。
如果此时Producer意识到消息发送失败并尝试再次发送消息此时Broker中就可能会出现两条内容相同并且Message ID也相同的消息那么后续Consumer就一定会消费两次该消息。
2.2消费时消息重复
消息已投递到Consumer并完成业务处理当Consumer给Broker反馈应答时网络闪断Broker没有接收到消费成功响应。
为了保证消息至少被消费一次的原则Broker将在网络恢复后再次尝试投递之前已被处理过的消息。
此时消费者就会收到与之前处理过的内容相同、Message ID也相同的消息。
2.3Rebalance时消息重复
当Consumer Group中的Consumer数量发生变化时或其订阅的Topic的Queue数量发生变化时会触发Rebalance
此时Consumer可能会收到曾经被消费过的消息。
3通用解决方案 重复消费没有办法剔除但是要保证重复的消费幂等 - 所以解决方案都是围绕保证幂等性的 3.1两要素(幂等令牌 唯一性处理)
幂等解决方案的设计中涉及到两项要素幂等令牌 唯一性处理。只要充分利用好这两要素就可以设计出好的幂等解决方案。
幂等令牌是生产者和消费者两者中的既定协议通常指具备唯一业务标识的字符串。 例如订单号、流水号。一般由Producer随着消息一同发送来的。 唯一性处理服务端通过采用一定的算法策略保证同一个业务逻辑不会被重复执行成功多次。 例如对同一笔订单的多次支付操作只会成功一次。
3.2通用解决方案
对于常见的系统幂等性操作的通用性解决方案是
首先通过缓存去重。在缓存中如果已经存在了某幂等令牌则说明本次操作是重复性操作若缓存没有命中则进入下一步。在唯一性处理之前先在数据库中查询幂等令牌作为索引的数据是否存在。若存在则说明本次操作为重复性操作若不存在则进入下一步。在同一事务中完成三项操作唯一性处理后将幂等令牌写入到缓存并将幂等令牌作为唯一索引的数据写入到DB中。 第 1 步已经判断过是否是重复性操作了为什么第 2 步还要再次判断 一般缓存中的数据是具有有效期的。缓存中数据的有效期一旦过期就是发生缓存穿透使请求直接就到达了DBMS。 3.3解决方案举例
以支付场景为例
当支付请求到达后首先在Redis缓存中却获取key为支付流水号的缓存value。 若value不空则说明本次支付是重复操作业务系统直接返回调用侧重复支付标识若value为空则进入下一步操作 到DBMS中根据支付流水号查询是否存在相应实例。 若存在则说明本次支付是重复操作业务系统直接返回调用侧重复支付标识若不存在则说明本次操作是首次操作进入下一步完成唯一性处理 在分布式事务中完成三项操作 完成支付任务将当前支付流水号作为key任意字符串作为value通过set(key, value, expireTime)将数据写入到Redis缓存将当前支付流水号作为主键与其它相关数据共同写入到DBMS
3.4消息幂等的实现
消费幂等的解决方案很简单为消息指定不会重复的唯一标识。因为Message ID有可能出现重复的情况所以真正安全的幂等处理不建议以Message ID作为处理依据。
最好的方式是以业务唯一标识作为幂等处理的关键依据而业务的唯一标识可以通过消息Key设置。
以支付场景为例可以将消息的Key设置为订单号作为幂等处理的依据。具体代码示例如下
Message message new Message();
message.setKey(ORDERID_100);
SendResult sendResult producer.send(message);消费者收到消息时可以根据消息的Key即订单号来实现消费幂等
consumer.registerMessageListener(new MessageListenerConcurrently() {Overridepublic ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context ) {for(MessageExt msg : msgs) {String key msg.getKeys();// 根据业务唯一标识Key做幂等处理// ......}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
});
RocketMQ能够保证消息不丢失但不能保证消息不重复
二消息堆积问题 消息处理流程中如果Consumer的消费速度跟不上Producer的发送速度MQ中未处理的消息会越来越多这部分消息就被称为堆积消息。
消息出现堆积进而会造成消息的消费延迟。以下场景需要重点关注消息堆积和消费延迟问题
业务系统上下游能力不匹配造成的持续堆积且无法自行恢复。业务系统对消息的消费实时性要求较高即使是短暂的堆积造成的消费延迟也无法接受。
1产生原因分析 Consumer使用长轮询Pull模式消费消息时分为以下两个阶段
1.1消息拉取
consumer通过长轮询Pull模式批量拉取的方式从服务端获取消息将拉取的消息缓存到本地缓冲队列中
对于拉取式消费在内网环境下会有很高的吞吐量所以这一阶段一般不会是消息堆积的瓶颈 一个单线程分区的低规格主机其可达到几万的TPS如果是多个分区的多个线程可以轻松达到几十万的TPS 1.2消息消费
Consumer将本地缓存的消息提交到消费线程中使用业务消费逻辑对消息进行处理处理完毕后获取到一个结果。这是真正的消息消费过程。
此时Consumer的消费能力就完全依赖于消息的消费耗时和消费并发度了。
如果由于业务处理逻辑复杂等原因导致处理单条消息的耗时较长则整体的消息吞吐量肯定不会高此时就会导致Consumer本地缓冲队列达到上限停止从服务端拉取消息。
所以消息堆积的主要瓶颈在于客户端的消费能力而消费能力由消费耗时和消费并发度决定。
注意消费耗时的优先级要高于消费并发度。即在保证了消费耗时的合理性前提下再考虑消费并发度问题。
2消费耗时
影响消息处理时长的主要因素是代码逻辑。而代码逻辑中可能会影响处理时长代码主要有两种类型CPU内部计算型代码和外部I/O操作型代码。
通常情况下代码中如果没有复杂的递归和循环的话内部计算耗时相对外部I/O操作来说几乎可以忽略。所以外部IO型代码是影响消息处理时长的主要症结所在。 外部IO操作型代码举例 读写外部数据库例如对远程MySQL的访问读写外部缓存系统例如对远程Redis的访问下游系统调用例如Dubbo的RPC远程调用Spring Cloud的对下游系统的Http接口调用 关于下游系统调用逻辑需要进行提前梳理掌握每个调用操作预期的耗时这样做是为了能够判断消费逻辑中IO操作的耗时是否合理。 通常消息堆积是由于下游系统出现了服务异常或达到了DBMS容量限制导致消费耗时增加。 服务异常并不仅仅是系统中出现的类似 500 这样的代码错误而可能是更加隐蔽的问题。例如网络带宽问题。 达到了DBMS容量限制其也会引发消息的消费耗时增加。 3消费并发度
一般情况下消费者端的消费并发度由单节点线程数和节点数量共同决定其值为value 单节点线程数*节点数量。
通常需要优先调整单节点的线程数若单机硬件资源达到了上限则需要通过横向扩展来提高消费并发度。 单节点线程数即单个Consumer所包含的线程数量 节点数量即Consumer Group所包含的Consumer数量 对于普通消息、延时消息及事务消息并发度计算都是单节点线程数*节点数量。但对于顺序消息则是不同的。 顺序消息的消费并发度等于Topic的Queue分区数量。
全局顺序消息
该类型消息的Topic只有一个Queue分区。其可以保证该Topic的所有消息被顺序消费。为了保证这个全局顺序性Consumer Group中在同一时刻只能有一个Consumer的一个线程进行消费。所以其并发度为 1 。
分区顺序消息
该类型消息的Topic有多个Queue分区。其仅可以保证该Topic的每个Queue分区中的消息被顺序消费不能保证整个Topic中消息的顺序消费。为了保证这个分区顺序性每个Queue分区中的消息在Consumer Group中的同一时刻只能有一个Consumer的一个线程进行消费。在同一时刻最多会出现多个Queue分蘖有多个Consumer的多个线程并行消费。所以其并发度为Topic的分区数量。
4单机线程的计算
对于一台主机中线程池中线程数的设置需要谨慎不能盲目直接调大线程数设置过大的线程数反而会带来大量的线程切换的开销。
理想环境下单节点的最优线程数计算模型为C *T1 T2/ T1。
CCPU内核数T1CPU内部逻辑计算耗时T2外部IO操作耗时 最优线程数 C *T1 T2/ T1 C * T1/T1 C * T2/T1 C C * T2/T1 注意该计算出的数值是理想状态下的理论数据在生产环境中不建议直接使用。而是根据当前环境先设置一个比该值小的数值然后观察其压测效果然后再根据效果逐步调大线程数直至找到在该环境中性能最佳时的值。 5如何避免消息堆积和消费延迟
为了避免在业务使用时出现非预期的消息堆积和消费延迟问题需要在前期设计阶段对整个业务逻辑进行完善的排查和梳理。
其中最重要的就是梳理消息的消费耗时和设置消息消费的并发度。
5.1梳理消息的消费耗时
通过压测获取消息的消费耗时并对耗时较高的操作的代码逻辑进行分析。梳理消息的消费耗时需要关注以下信息
消息消费逻辑的计算复杂度是否过高代码是否存在无限循环和递归等缺陷。消息消费逻辑中的I/O操作是否是必须的能否用本地缓存等方案规避。消费逻辑中的复杂耗时的操作是否可以做异步化处理。如果可以是否会造成逻辑错乱。
5.2设置消费并发度
对于消息消费并发度的计算可以通过以下两步实施
逐步调大单个Consumer节点的线程数并观测节点的系统指标得到单个节点最优的消费线程数和消息吞吐量。根据上下游链路的流量峰值计算出需要设置的节点数 节点数 流量峰值 / 单个节点消息吞吐量 三消息清理问题
消息被消费过后会被清理掉吗不会的。
消息是被顺序存储在commitlog文件的且消息大小不定长所以消息的清理是不可能以消息为单位进行清理的
消息是以commitlog文件为单位进行清理的。否则会急剧下降清理效率并实现逻辑复杂。
commitlog文件存在一个过期时间默认 72 小时
除了用户手动清理外在以下情况下也会被自动清理无论文件中的消息是否被消费过
文件过期且到达清理时间点默认为凌晨 4 点后自动清理过期文件文件过期且磁盘空间占用率已达过期清理警戒线默认75%后无论是否达到清理时间点都会自动清理过期文件磁盘占用率达到清理警戒线默认85%后开始按照设定好的规则清理文件无论是否过期。默认会从最老的文件开始清理磁盘占用率达到系统危险警戒线默认90%后Broker将拒绝消息写入
需要注意以下几点
对于RocketMQ系统来说删除一个1G大小的文件是一个压力巨大的IO操作。 在删除过程中系统性能会骤然下降。所以其默认清理时间点为凌晨 4 点访问量最小的时间。也正因如果我们要保障磁盘空间的空闲率不要使系统出现在其它时间点删除commitlog文件的情况。 官方建议RocketMQ服务的Linux文件系统采用ext4。因为对于文件删除操作ext4要比ext3性能更好
四消息过滤问题
消息者在进行消息订阅时除了可以指定要订阅消息的Topic外还可以对指定Topic中的消息根据指定条件进行过滤即可以订阅比Topic更加细粒度的消息类型。
对于指定Topic消息的过滤有两种过滤方式Tag过滤与SQL过滤。
1Tag方式过滤-简单
通过consumer的subscribe()方法指定要订阅消息的Tag。如果订阅多个Tag的消息Tag间使用或运算符(双竖线||)连接。
DefaultMQPushConsumer consumer new DefaultMQPushConsumer(CID_EXAMPLE);
consumer.subscribe(TOPIC, TAGA || TAGB || TAGC);2SQL过滤-功能强大
SQL过滤是一种通过特定表达式对事先埋入到消息中的用户属性进行筛选过滤的方式。
通过SQL过滤可以实现对消息的复杂过滤。不过只有使用PUSH模式的消费者才能使用SQL过滤。
SQL过滤表达式中支持多种常量类型与运算符。
支持的常量类型
数值比如 123 3.1415字符必须用单引号包裹起来比如‘abc’布尔TRUE 或 FALSENULL特殊的常量表示空
支持的运算符有
数值比较BETWEEN字符比较IN逻辑运算 ANDORNOTNULL判断IS NULL 或者 IS NOT NULL
默认情况下Broker没有开启消息的SQL过滤功能需要在Broker加载的配置文件中添加如下属性以开启该功能
enablePropertyFilter true在启动Broker时需要指定这个修改过的配置文件。
例如对于单机Broker的启动其修改的配置文件是conf/broker.conf启动时使用如下命令
sh bin/mqbroker -n localhost:9876 -c conf/broker.conf 3代码举例
3.1Tag方式进行过滤例子 定义Tag过滤Producer public class FilterByTagProducer {public static void main(String[] args) throws Exception {DefaultMQProducer producer new DefaultMQProducer(pg);producer.setNamesrvAddr(rocketmqOS:9876);producer.start();String[] tags {myTagA,myTagB,myTagC};for (int i 0 ; i 10 ; i) {byte[] body (Hi, i).getBytes();String tag tags[i % tags.length];// 指定tagMessage msg new Message(myTopic, tag, body);SendResult sendResult producer.send(msg);System.out.println(sendResult);}producer.shutdown();}
}定义Tag过滤Consumer public class FilterByTagConsumer {public static void main(String[] args) throws Exception {DefaultMQPushConsumer consumer new DefaultMQPushConsumer(pg);consumer.setNamesrvAddr(rocketmqOS:9876); // 设置NameServerconsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);// 消费者订阅指定的tag, 多个tag用||分割consumer.subscribe(myTopic, myTagA || myTagB); // 订阅主题和tag只有tagA或者tagB才进行消费// 定义异步并发监听准备消费consumer.registerMessageListener(new MessageListenerConcurrently() {Overridepublic ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs,ConsumeConcurrentlyContext context) {for (MessageExt me:msgs){System.out.println(me);}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.println(Consumer Started);}
}3.2SQL方式进行过滤例子 定义SQL过滤Producer public class FilterBySQLProducer {public static void main(String[] args) throws Exception {DefaultMQProducer producer new DefaultMQProducer(pg);producer.setNamesrvAddr(rocketmqOS:9876);producer.start();for (int i 0 ; i 10 ; i) {try {byte[] body (Hi, i).getBytes();Message msg new Message(myTopic, myTag, body);// 自己声明一个属性并赋值到时候sqmsg.putUserProperty(age, i );SendResult sendResult producer.send(msg);System.out.println(sendResult);} catch (Exception e) {e.printStackTrace();}}producer.shutdown();}
}定义SQL过滤Consumer public class FilterBySQLConsumer {public static void main(String[] args) throws Exception {DefaultMQPushConsumer consumer new DefaultMQPushConsumer(pg);consumer.setNamesrvAddr(rocketmqOS:9876);consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);// 注意下面这句consumer.subscribe(myTopic, MessageSelector.bySql(age between 0 and 6));// 监听并consumer.registerMessageListener(new MessageListenerConcurrently() {Overridepublic ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context) {for (MessageExt me:msgs){System.out.println(me);}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.println(Consumer Started);}
}五消息重试问题
1消息发送的重试机制
Producer对发送失败的消息进行重新发送的机制称为消息发送重试机制也称为消息重投机制。 注意事项 生产者在发送消息时若采用同步或异步发送方式发送失败会重试但oneway消息发送方式发送失败是没有重试机制的只有普通消息具有发送重试机制顺序消息是没有的消息重投机制可以保证消息尽可能发送成功、不丢失但可能会造成消息重复。消息重复在RocketMQ中是无法避免的问题消息发送重试有三种策略可以选择同步发送失败策略、异步发送失败策略、消息刷盘失败策略
1.1同步发送失败策略
对于普通消息消息发送默认采用round-robin策略来选择所发送到的队列。
如果发送失败默认重试 2 次。
重试时是不会选择上次发送失败的Broker而是选择其它Broker。若只有一个Broker其也只能发送到该Broker但其会尽量发送到该Broker上的其它Queue。
// 创建一个producer参数为Producer Group名称
DefaultMQProducer producer new DefaultMQProducer(pg);
// 指定nameServer地址
producer.setNamesrvAddr(rocketmqOS:9876);
// 设置同步发送失败时重试发送的次数默认为 2 次
producer.setRetryTimesWhenSendFailed( 3 );
// 设置发送超时时限为5s默认3s
producer.setSendMsgTimeout( 5000 );同时Broker还具有失败隔离功能使Producer尽量选择未发生过发送失败的Broker作为目标Broker。
这样做可以保证其它消息尽量不发送到问题Broker为了提升消息发送效率降低消息发送耗时。 思考让我们自己实现失败隔离功能如何来做 方案一Producer中维护某JUC的Map集合其key是发生失败的时间戳value为Broker实例。Producer中还维护着一个Set集合其中存放着所有未发生发送异常的Broker实例。选择目标Broker是从该Set集合中选择的。再定义一个定时任务定期从Map集合中将长期未发生发送异常的Broker清理出去并添加到Set集合。方案二为Producer中的Broker实例添加一个标识例如是一个AtomicBoolean属性。只要该Broker上发生过发送异常就将其置为true。选择目标Broker就是选择该属性值为false的Broker。再定义一个定时任务定期将Broker的该属性置为false。方案三为Producer中的Broker实例添加一个标识例如是一个AtomicLong属性。只要该Broker上发生过发送异常就使其值增一。选择目标Broker就是选择该属性值最小的Broker。若该值相同采用轮询方式选择。 如果超过重试次数则抛出异常由Producer去保证消息不丢。
当然当生产者出现RemotingException、MQClientException和MQBrokerException时Producer会自动重投消息。
1.2异步发送失败策略
异步发送失败重试时异步重试不会选择其他broker仅在同一个broker上做重试所以该策略无法保证消息不丢。
DefaultMQProducer producer new DefaultMQProducer(pg);
producer.setNamesrvAddr(rocketmqOS:9876);
// 指定异步发送失败后不进行重试发送
producer.setRetryTimesWhenSendAsyncFailed( 0 );1.3消息刷盘失败策略
消息刷盘超时或slave不可用slave在做数据同步时向master返回状态不是SEND_OK时默认是不会将消息尝试发送到其他Broker的。
对于重要消息可以通过在Broker的配置文件设置retryAnotherBrokerWhenNotStoreOK true来开启。
2消费重试
2.1顺序消息的消费重试
对于顺序消息当Consumer消费消息失败后为了保证消息的顺序性其会自动不断地进行消息重试直到消费成功。
消费重试默认间隔时间为 1000 毫秒。重试期间应用会出现消息消费被阻塞的情况。
DefaultMQPushConsumer consumer new DefaultMQPushConsumer(cg);
// 顺序消息消费失败的消费重试时间间隔单位毫秒默认为 1000 其取值范围为[10,30000]
consumer.setSuspendCurrentQueueTimeMillis( 100 );由于对顺序消息的重试是无休止的不间断的直至消费成功 ⚠️ 所以对于顺序消息的消费务必要保证应用能够及时监控并处理消费失败的情况避免消费被永久性阻塞。 ⚠️ 顺序消息没有发送失败重试机制但具有消费失败重试机制 2.2无序消息的消费重试
对于无序消息普通消息、延时消息、事务消息当Consumer消费消息失败时可以通过设置返回状态达到消息重试的效果。
不过需要注意无序消息的重试只对集群消费方式生效广播消费方式不提供失败重试特性。
即对于广播消费消费失败后失败消息不再重试继续消费后续消息。
2.3消息重试次数和间隔
对于无序消息集群消费下的重试消费每条消息默认最多重试 16 次但每次重试的间隔时间是不同的会逐渐变长。
每次重试的间隔时间如下表。
重试次数与上次重试的间隔时间重试次数与上次重试的间隔时间110秒97分钟230108 分钟31分钟119 分钟42分钟1210分钟53分钟1320分钟64分钟1430分钟75分钟151小时86分钟162 小时 若一条消息在一直消费失败的前提下将会在正常消费后的第 4 小时 46 分后进行第 16 次重试。 若仍然失败则将消息投递到死信队列 修改消费重试次数 DefaultMQPushConsumer consumer new DefaultMQPushConsumer(cg);
// 修改消费重试次数
consumer.setMaxReconsumeTimes( 10 );对于修改过的重试次数将按照以下策略执行 若修改值 16 则按照指定间隔进行重试若修改值 16 则超过 16 次的重试时间间隔均为 2 小时 对于Consumer Group 若仅修改了一个Consumer的消费重试次数则会应用到该Group中所有其它Consumer实例。 若出现多个Consumer均做了修改的情况则采用覆盖方式生效。即最后被修改的值会覆盖前面设置的值。 2.4重试队列
对于需要重试消费的消息并不是Consumer在等待了指定时长后再次去拉取原来的消息进行消费而是将这些需要重试消费的消息放入到了一个特殊Topic的队列中而后进行再次消费的。这个特殊的队列就是重试队列。
当出现需要进行重试消费的消息时Broker会为每个消费组都设置一个Topic名称为%RETRY%consumerGroupconsumerGroup的重试队列。 这个重试队列是针对消息才组的而不是针对每个Topic设置的只有当出现需要进行重试消费的消息时才会为该消费者组创建重试队列 注意消费重试的时间间隔与延时消费的延时等级十分相似 除了没有延时等级的前两个时间外其它的时间都是相同的 Broker对于重试消息的处理是通过延时消息实现的。
先将消息保存到SCHEDULE_TOPIC_XXXX延迟队列中延迟时间到后会将消息投递到%RETRY%consumerGroupconsumerGroup重试队列中。
2.5消息重试/不重试配置方式
2.5.1重试配置
集群消费方式下消息消费失败后若希望消费重试则需要在消息监听器接口的实现中明确进行如下三种方式之一的配置
返回ConsumeConcurrentlyStatus.RECONSUME_LATER推荐返回Null抛出异常
Override
public ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context) {try {...} catch (Throwable e) {// 以下三种方式都可以引发消息重试return ConsumeConcurrentlyStatus.RECONSUME_LATER;// return null;// throw new RuntimeException(消费异常);}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}2.5.2不重试配置
在捕获到异常后同样也返回与消费成功后的相同的结果即ConsumeConcurrentlyStatus.CONSUME_SUCCESS则不进行消费重试。
Override
public ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context) {try {...} catch (Throwable e) {// 也return成功就不进行重试return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}2.6死信队列
当一条消息初次消费失败消息队列会自动进行消费重试达到最大重试次数后若消费依然失败则表明消费者在正常情况下无法正确地消费该消息此时消息队列不会立刻将消息丢弃而是将其发送到该消费者对应的特殊队列中。这个队列就是死信队列Dead-Letter QueueDLQ
死信队列其中的消息称为死信消息Dead-Letter MessageDLM。 死信队列是用于处理无法被正常消费的消息的。 2.6.1死信队列特征
死信队列中的消息不会再被消费者正常消费即DLQ对于消费者是不可见的死信存储有效期与正常消息相同均为 3 天commitlog文件的过期时间 3 天后会被自动删除死信队列就是一个特殊的Topic名称为%DLQ%consumerGroupconsumerGroup即每个消费者组都有一个死信队列如果一个消费者组未产生死信消息则不会为其创建相应的死信队列一个死信队列包含了对应Group ID产生的所有死信消息不论该消息属于哪个Topic。
2.6.2查询死信消息
消息队列RocketMQ版提供的查询死信消息的方式对比如下表所示。
查询方式查询条件查询类别说明按Group ID查询Group ID 时间段范围查询根据Group ID和时间范围批量获取符合条件的所有消息查询量大不易匹配。按Message ID查询Group ID Message ID精确查询根据Group ID和Message ID可以精确定位任意一条消息。 登录消息队列RocketMQ版控制台在左侧导航栏单击实例列表。 在顶部菜单栏选择地域如华东1杭州然后在实例列表中单击目标实例名称。 在左侧导航栏单击死信队列然后在死信队列页面选择以下任一方式查询死信消息
按 Group 查询 - 根据Group ID和死信消息产生的时间范围批量查询该Group ID在某段时间内产生的所有死信消息。页面中会显示所有符合筛选条件的死信消息 按Message ID查询消息属于精确查询。您可以根据Group ID与Message ID精确查询到任意一条消息。页面中会显示所有符合筛选条件的死信消息 2.6.3死信队列处理
实际上当一条消息进入死信队列就意味着系统中某些地方出现了问题从而导致消费者无法正常消费该消息比如代码中原本就存在Bug。
因此需要开发人员进行特殊处理。最关键的步骤是要排查可疑因素解决代码中可能存在的Bug然后再将原来的死信消息再次进行投递消费
六如何保证消息不丢失
1Producer如何保证消息不丢失
生产者发送消息之后给生产者一个确定的通知这个消息在Broker端是否写入完成了。
//异步发送不需要broker确认效率高但是会有丢消息的可能。
producer.sendOneway(msg);
//同步发送生产者等待Broker确认。消息最安全但效率最低
SendResult sendResult producer.send(msg,20*1000);
//异步发送生产者另起一个线程等待broker确认收到Broker确认之后直接触发回调方法。消息安全和效率之间比较均衡但是会加大客户端的负担。
producer.send(msg,new SendCallback(){Overridepublic void onSuccess(SendResult sendResult){//do something}Overridepublic void onException(Throwable e){//do something}
})与之类似的Kafka同样也提供了这种同步和异步的发送消息机制
//直接send发送消息返回的是一个Future。这就相当于是异步调用
FutureRecordMetadata future producer.send(record);
//调用future的get方法才会世纪获取到发送的结果生产者收到这个结果后就可以知道消息是否成功发送到broker了。这个过程就变成了一个同步的过程
RecordMetadata recordMetadata producer.send(record).get();而在RabbitMQ中则是提供了一个Publisher Confirms生产者确认机制。
其思路也是Publisher收到Broker的响应后再发出对应的回调方法。
//获取channel
Channel ch ...;
//添加两个回调一个处理ack响应一个处理nack响应
ch.addConfirmListener(ConfirmCallback ackCallback,ConfirmCallback nackCallback)2事务消息保证消息不丢失 先定义本地事务监听器
package com.mytest.mqdemo.producer;import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;/*** p* 功能描述* /p** author cui haida* date 2024/03/24/17:52*/
public class TransactionListener implements org.apache.rocketmq.client.producer.TransactionListener {/*** 执行本地事务* param message 消息体* param o 参数* return 执行本地事务的结果*/Overridepublic LocalTransactionState executeLocalTransaction(Message message, Object o) {// todo: 执行本地事务这里这是模拟String tags message.getTags();if (StringUtils.contains(A, tags)) {return LocalTransactionState.COMMIT_MESSAGE;}if (StringUtils.contains(B, tags)) {return LocalTransactionState.ROLLBACK_MESSAGE;} else {return LocalTransactionState.UNKNOW;}}/*** 回查本地事务* param message 待回查的本地事务信息* return 回查结果*/Overridepublic LocalTransactionState checkLocalTransaction(MessageExt message) {String tags message.getTags();if (StringUtils.contains(C, tags)) {return LocalTransactionState.COMMIT_MESSAGE;}if (StringUtils.contains(D, tags)) {return LocalTransactionState.ROLLBACK_MESSAGE;} else {return LocalTransactionState.UNKNOW;}}
}生产者指定本地事务监听器并进行消息的发送
package com.mytest.mqdemo.producer;import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;import java.nio.charset.StandardCharsets;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;/*** p* 功能描述* /p** author cui haida* date 2024/03/24/17:47*/
public class TransactionProducer {public static void main(String[] args) throws Exception {// 注意这里使用的是事务消息生产者TransactionMQProducer transactionProducer new TransactionMQProducer(transaction_producer);transactionProducer.setNamesrvAddr(192.168.111.128:9876);// 异步提交提高性能ThreadPoolExecutor threadPoolExecutor new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, // 4大基本参数new ArrayBlockingQueue(2000), // 队列类型和大小new ThreadFactory() { // 线程工厂指定Overridepublic Thread newThread(Runnable r) {Thread thread new Thread(r);thread.setName(test-for-trans);return thread;}});transactionProducer.setExecutorService(threadPoolExecutor);// 创建一个本地事务的监听器transactionProducer.setTransactionListener(new TransactionListener());transactionProducer.start();String[] tags new String[]{A, B, C, D, E};for (int i 0; i 10; i) {String body (tags[i % tags.length] trans_message);Message message new Message(trans, tags[i % tags.length], body.getBytes(StandardCharsets.UTF_8));// 发送TransactionSendResult transactionSendResult transactionProducer.sendMessageInTransaction(message, null);System.out.println(消息发送成功_ transactionSendResult);Thread.sleep(10);}Thread.sleep(10000);transactionProducer.shutdown();}
}定义消费者信息
因为半消息和回查对于消费者无感知所以消费者就是最基本的实现
package com.mytest.mqdemo.customer;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;import java.nio.charset.StandardCharsets;/*** author cuihaida* p* 简单消费* /p*/
public class PushConsumer {public static void main(String[] args) throws MQClientException {// 消费的推拉模式// 拉模式pull - 消费者主动去Broker上拉取消息// 推模式push - 消费者等待Broker将消息推送过来// 定义一个push消费者String consumerGroup cg;DefaultMQPushConsumer consumer new DefaultMQPushConsumer(consumerGroup);// 指定nameServerconsumer.setNamesrvAddr(192.168.111.129:9876);// 指定消费topic与tagString consumerTopic someTopic;String subTag *; // 当前主题下的任意标签消息consumer.subscribe(consumerTopic, subTag);// 指定采用“广播模式”进行消费默认为“集群模式”// consumer.setMessageModel(MessageModel.BROADCASTING);// 注册消息监听器// 一旦broker中有了其订阅的消息就会触发该方法的执行其返回值为当前consumer消费的状态// MessageListener有两个实现类// -- MessageListenerConcurrently - 并发消费// -- MessageListenerOrderly - 顺序消费// 第一个参数是消息列表第二个参数是上下文consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) - {// 逐条消费消息for (MessageExt msg : messages) {try {String msgBody new String(msg.getBody(), StandardCharsets.UTF_8);System.out.println(msgBody);// todo: processMessage(msgBody);// 手动ACK返回消费状态消费成功return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;} catch (Exception e) {System.out.println(消息消费异常);// todo: 打印堆栈信息异常处理// 手动ACK返回消费状态需要重新消费return ConsumeConcurrentlyStatus.RECONSUME_LATER;}}// 返回消费状态消费成功return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});// 开启消费者消费consumer.start();System.out.println( Consumer Started );}
}3broker保证消息不丢失
Producer把消息发送到Broker上了之后Broker是不是能保证消息不丢失呢这里有一个核心问题——PageCache缓存
数据会优先写入缓存然后过一段时间再写图到磁盘。但是缓存中的数据有个特点就是断点即丢失所以如果服务器发生非正常断电内存中的数据还没有写入磁盘这是就会造成消息丢失。
以Linux为例用户态就是应用程序不管是什么应用程序想要写入磁盘文件时都只能调用操作系统提供的write系统调用申请写磁盘。至于消息如何经过PageCache再写入到磁盘中这个过程就是在内核态执行的也就是操作系统自己执行的应用程序无法干预。这个过程中应用系统唯一能够干预的就是调用操作系统提供的sync系统调用申请一次刷盘操作主动将PageCache中的数据写入到磁盘。
4Broker主从同步如何保证不丢失
对于Broker来说通常Slaver的作用就是做一个数据备份当Broker服务宕机时甚至是磁盘坏了时可以从Slaver上获取数据记录。
但是如果主从同步失败了那么Broker的这一层保证就会失效。因此主从同步也有可能造成消息的丢失。
普通集群 - 指定角色各司其职Dledger高可用集群自行选举多数同意
5Consumer如何保证消费不丢失
几乎所有的MQ产品都设置了消费者消费者状态确认机制。也就是消费者处理完消息之后需要给Broker一个响应表示消息被正常处理了。
如果Broker没有拿到这个响应不管是因为Consumer没有拿到还是Consumer处理完消息后咩有给出响应Broker都会认为消息没有处理成功。之后Broker就会向Consumer重复投递这些没有处理成功的消息。
6如果MQ服务全挂了如何保证不丢失
针对这种情况通常做法是设计一个降级缓存。Producer往MQ发消息失败了就往降级缓存中写然后依然正常去进行后续的业务。
此时再启动一个线程不断尝试将降级缓存中的数据往MQ中发送。这样至少当MQ服务恢复之后这些消息可以尽快进入到MQ中继续往下游Consumer推送而不至于造成消息丢失。
