其他分享
首页 > 其他分享> > 22第四章:07_消息发送重试机制

22第四章:07_消息发送重试机制

作者:互联网

一、消息发送重试机制说明

Producer 对发送失败的消息进行重新发送的机制,称为消息发送重试机制,也称为消息重投机制。

对于消息重投,需要注意以下几点:

二、同步发送失败策略

对于普通消息,消息发送默认采用 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,为了提升消息发送效率,降低消息发送耗时。

思考:让我们自己实现 失败隔离 功能,如何来做?

1)方案一:Producer 中维护某 JUC 的 Map 集合,其 key 是发生失败的时间戳,value 为 Broker 实例。Producer 中还维护着一个 Set 集合,其中存放着所有未发生发送异常的 Broker 实例。选择目标 Broker 是从该 Set 集合中选择的。再定义一个定时任务,定期从 Map 集合中将长期未发生发送异常的 Broker 清理出去,并添加到 Set 集合。

2)方案二:为 Producer 中的 Broker 实例添加一个标识,例如是一个 AtomicBoolean 属性。只要该 Broker 上发生过发送异常,就将其置为 true。选择目标 Broker 就是选择该属性值为 false 的 Broker。再定义一个定时任务,定期将 Broker 的该属性置为 false。

3)方案三:为 Producer 中的 Broker 实例添加一个标识,例如是一个 AtomicLong 属性。只要该 Broker 上发生过发送异常,就使其值增一。选择目标 Broker 就是选择该属性值最小的 Broker。若该值相同,采用轮询方式选择。

如果超过重试次数,则抛出异常,由 Producer 去保证消息不丢。当然当生产者出现 RemotingException、MQClientException 和 MQBrokerException 时,Producer 会自动重投消息。

三、 异步发送失败策略

异步发送失败重试时,异步重试不会选择其他 broker,仅在同一个 broker 上做重试,所以该策略无法保证消息不丢。

DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
// 指定异步发送失败后不进行重试发送
producer.setRetryTimesWhenSendAsyncFailed(0)

四、消息刷盘失败策略

消息刷盘超时(Master 或 Slave)或 slave 不可用(slave 在做数据同步时向 master 返回状态不是 SEND_OK)时,默认是不会将消息尝试发送到其他 Broker 的。不过,对于重要消息可以通过在 Broker 的配置文件设置 retryAnotherBrokerWhenNotStoreOK 属性为 true 来开启。

标签:发送,07,22,Broker,重试,失败,消息,Producer
来源: https://www.cnblogs.com/niujifei/p/16586484.html