SpringBoot整合RocketMQ发送顺序消息

1. 有序性分类
根据有序范围的不同，可以分为两种消息的有序性：分区有序和全局有序
分区有序
有多个Queue参与，其仅可保证在该Queue分区队列上的消息顺序，称为分区有序
在定义Producer时我们可以指定消息队列选择器，而这个选择器是我们自己实现了MessageQueueSelector接口定义的。在定义选择器的选择算法时，一般需要使用选择key。这个选择key可以是消息key也可以是其它数据。但无论谁做选择key，都不能重复，都是唯一的。

一般性的选择算法是，让选择key（或其hash值）与该Topic所包含的Queue的数量取模，其结果即为选择出的Queue的QueueId。

取模算法存在一个问题：不同选择key与Queue数量取模结果可能会是相同的，即不同选择key的消息可能会出现在相同的Queue，即同一个Consuemr可能会消费到不同选择key的消息。这个问题如何解决？一般性的作法是，从消息中获取到选择key，对其进行判断。若是当前Consumer需要消费的消息，则直接消费，否则，什么也不做。这种做法要求选择key要能够随着消息一起被Consumer获取到。此时使用消息key作为选择key是比较好的做法。
全局有序
当发送和消费参与的Queue只有一个时所保证的有序是整个Topic中消息的顺序，称为全局有序
在创建Topic时指定Queue的数量。有三种指定方式：

在代码中创建Producer时，可以指定其自动创建的Topic的Queue数量
在RocketMQ可视化控制台中手动创建Topic时指定Queue数量
使用mqadmin命令手动创建Topic时指定Queue数量
2. 生产者业务接口

public interface OrderMessageService {

    /**
     * 发送同步顺序消息
     * @param id
     * @param message
     */
    void sendSyncOrderMessage(String id, String message);

    /**
     * 发送异步顺序消息
     * @param id
     * @param message
     */
    void sendAsyncOrderMessage(String id, String message);

    /**
     * 发送单向顺序消息
     * @param id
     * @param message
     */
    void sendOnewayOrderMessage(String id, String message);
}

3. 生产者业务接口实现类

@Service
public class OrderMessageServiceImpl implements OrderMessageService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    private static final Logger logger = LoggerFactory.getLogger(OrderMessageServiceImpl.class);

    @Override
    public void sendSyncOrderMessage(String id, String message) {
        Message<String> strMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, id).build();
        rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> list, org.apache.rocketmq.common.message.Message message, Object obj) {
                Integer uid = Integer.valueOf(String.valueOf(obj));
                int index = uid % list.size();
                return list.get(index);
            }
        });
        SendResult result = rocketMQTemplate.syncSendOrderly("order-message-topic:sync-tags", strMessage, id);
        if (result.getSendStatus() == SendStatus.SEND_OK) {
            logger.info("发送同步顺序消息成功!");
        } else {
            logger.error("发送同步顺序消息失败!消息ID为:{}", result.getMsgId());
        }
    }

    @Override
    public void sendAsyncOrderMessage(String id, String message) {
        Message<String> strMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, id).build();
        rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> list, org.apache.rocketmq.common.message.Message message, Object obj) {
                Integer uid = (Integer) obj;
                int index = uid % list.size();
                return list.get(index);
            }
        });
        rocketMQTemplate.asyncSendOrderly("order-message-topic:async-tags", strMessage, id, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                    logger.info("发送异步顺序消息成功!消息ID为:{}", sendResult.getMsgId());
                }
            }
            @Override
            public void onException(Throwable throwable) {
                logger.info("发送异步顺序消息失败!失败原因为:{}", throwable.getMessage());
            }
        });
    }

    @Override
    public void sendOnewayOrderMessage(String id, String message) {
        Message<String> strMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, id).build();
        rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> list, org.apache.rocketmq.common.message.Message message, Object obj) {
                Integer uid = (Integer) obj;
                int index = uid % list.size();
                return list.get(index);
            }
        });
        rocketMQTemplate.sendOneWayOrderly("order-message-topic:oneway-tags", strMessage, id);
    }
}

4. 消费者类

@Component
@RocketMQMessageListener(topic = "order-message-topic", consumerGroup = "order-consumer-group", consumeMode = ConsumeMode.ORDERLY)
public class OrderMessageListener implements RocketMQListener<String> {

    private static final Logger logger = LoggerFactory.getLogger(OrderMessageListener.class);

    @Override
    public void onMessage(String message) {
        logger.info("接收到顺序消息为:{}", message);
    }
}

5. 测试

@Test
void orderMessage() {
    for (int i = 1; i < 5; i++) {
        orderMessageService.sendSyncOrderMessage(String.valueOf(i), "hello" + i);
    }
}