高性能消息中间件 RocketMQ（三）

6.2 负载均衡

RocketMQ中的负载均衡都在Client端完成，具体来说的话，主要可以分为Producer端发送消息时候的负载均衡和Consumer端订阅消息的负载均衡。

• 生产者的负载均衡：
  

如图所示，5 个队列可以部署在一台机器上，也可以分别部署在 5 台不同的机器上，发送消息通过轮询队列的方式 发送，每个队列接收平均的消息量。通过增加机器，可以水平扩展队列容量。 另外也可以自定义方式选择发往哪个队列。

• 消费者的负载均衡：
  

如图所示，如果有 5 个队列，2 个 consumer，那么第一个 Consumer 消费 3 个队列，第二consumer 消费 2 个队列。 这样即可达到平均消费的目的，可以水平扩展 Consumer 来提高消费能力。但是 Consumer 数量要小于等于队列数 量，如果 Consumer 超过队列数量，那么多余的Consumer 将不能消费消息 。

6.3 事务消息

Apache RocketMQ在4.3.0版中已经支持分布式事务消息，这里RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示：

事务消息发送步骤如下：

• 生产者将半事务消息发送至消息队列RocketMQ服务端(也就是broker)，半消息不能被消费，只有将半消息标记为正常的消息才能被消费。
  
• 消息队列RocketMQ服务端将消息持久化成功之后，向生产者返回Ack确认消息已经发送成功，此时消息为半事务消息。
  
• 生产者开始执行本地事务逻辑。
  
• 生产者根据本地事务执行结果向服务端提交二次确认结果（Commit或是Rollback），服务端收到确认结果后处理逻辑如下：

• 二次确认结果为Commit：服务端将半事务消息标记为可投递，并投递给消费者。
• 二次确认结果为Rollback：服务端不会将该消息投递给消费者，并按照如下逻辑进行回查处理：
• 在断网或者是生产者应用重启的特殊情况下，上图步骤4提交的二次确认最终未到达服务端，经过固定时间后，服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查。
• 生产者收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
• 生产者根据检查得到的本地事务的最终状态再次提交二次确认，服务端仍按照上图4步骤对半事务消息进行处理。

6.4 顺序消息

消息有序指的是按照消息的发送顺序来消费(FIFO)。RocketMQ可以保证消息有序，消息有序分为部分有序和全局有序。全局有序是指某个Topic下的所有消息都要保证顺序；部分顺序消息只要保证每一组消息被顺序消费即可。

如果要实现全局顺序消息，那么只能使用一个队列，一个生产者，这会严重影响性能。

因此，我们常说的顺序消息通常指部分顺序消息。

顺序消费的原理解析，在默认的情况下消息发送会采取轮询方式把消息发送到不同的分区队列；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。

6.5 消息重试

生产端重试

例如由于网络原因导致生产者发送消息到MQ失败，即发送端没有收到Broker的ACK，导致最终Consumer无法消费消息，此时RocketMQ会自动进行重试。

// 同步发送消息，如果5秒内没有发送成功，则重试3次
DefaultMQProducer producer = new DefaultMQProducer("DefaultProducer");
producer.setRetryTimesWhenSendFailed(3);
producer.send(msg, 5000L);

消费端重试

同样的，由于网络原因，Broker发送消息给消费者后，没有受到消费端的ACK响应，所以Broker又会尝试将消息重新发送给Consumer，在实际开发过程中，我们更应该考虑的是消费端的重试。消费端的消息重试可以分为顺序消息的重试以及无序消息的重试。

• 顺序消息重试
  对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 会自动不断进行消息重试（每次间隔时间为 1 秒），这时，应用会出现消息消费被阻塞的情况。因此，在使用顺序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生
  
• 无序消息重试
  对于无序消息（普通、定时、延时、事务消息），当消费者消费消息失败时，可以通过设置返回状态达到消息重试的结果。
  

最大重试次数

消息消费失败后，可被消息队列RocketMQ重复投递的最大次数。

TCP协议无序消息重试时间间隔：

消费失败后重新配置方式

需要在消息监听器接口的实现中明确进行配置（三种方式任选一种）：

• 返回 ConsumeConcurrentlyStatus.RECONSUME_LATER; （推荐）
• 返回 Null
• 抛出异常

6.6 延迟消息

Producer将消息发送到消息队列RocketMQ服务端，但并不期望立马投递这条消息，而是延迟一定时间后才投递到Consumer进行消费，该消息即延时消息。

消息生产和消费有时间窗口要求，例如在电商交易中超时未支付关闭订单的场景，在订单创建时会发送一条延时消息。这条消息将会在30分钟以后投递给消费者，消费者收到此消息后需要判断对应的订单是否已完成支付。如支付未完成，则关闭订单。如已完成支付则忽略。通过消息触发一些定时任务，例如在某一固定时间点向用户发送提醒消息。

定时消息会暂存在名为SCHEDULE_TOPIC_XXXX的topic中，并根据delayTimeLevel存入特定的queue，queueId = delayTimeLevel – 1，即一个queue只存相同延迟的消息，保证具有相同发送延迟的消息能够顺序消费。broker会调度地消费SCHEDULE_TOPIC_XXXX，将消息写入真实的topic。

6.7 消息查询

在实际开发中，经常需要查看MQ中消息的内容来排查问题。 生产者将消息发送到broker之后，broker会自动为消息创建一个messageID，RocketMQ提供了三种消息查询的方式，分别是按Message ID、Message Key以及Unique Key查询。

//返回结果
SendResult [
sendStatus=SEND_OK, 
msgId=C0A801030D4B18B4AAC247DE4A0D0000,
offsetMsgId=C0A8010300002A9F000000000007BEE9,
messageQueue=MessageQueue [topic=TopicA, brokerName=broker-a, queueId=0], 
queueOffset=0]

• 按MessageId查询消息
  Message Id 是消息发送后，在Broker端生成的，其包含了Broker的地址、偏移信息，并且会把Message Id作为结果的一部分返回。Message Id中属于精确匹配，代表唯一一条消息，查询效率更高。
  
• 按照Message Key查询消息
  消息的key是开发人员在发送消息之前自行指定的，通常把具有业务含义，区分度高的字段作为消息的key，如用户id，订单id等。
  
• 按照Unique Key查询消息
  除了开发人员指定的消息key，生产者在发送发送消息之前，会自动生成一个UNIQ_KEY，设置到消息的属性中，从逻辑上唯一代表一条消息。
  

消息在消息队列RocketMQ中存储的时间默认为3天（不建议修改），即只能查询从消息发送时间算起3天内的消息，三种查询方式的特点和对比如下表所述：