rocketmq顺序消费问题

场景举例

用户下单，付款，发货，完成，每一个步骤发送完成，发送mq消息到下游系统处理（消费客户端），同一个订单要保证顺序消费。

mq实现思路

选择mq顺序消息，实现思路，

生产者：发送消息的时候保证同一个订单的业务消息，发送到同一个队列；订单号对队列总和取模，找到指定队列

消费者：每一个消费线程可以保证只消费一个队列的消息，怎么保证的...可进一步探究

代码

生产者

/*** 顺序消息生产者*/
@Slf4j
public class OrderedProducer {private static List<Order> buildMsg() {List<Order> orders = new ArrayList<>();Order order1 = new Order();order1.setOrderNo("1");order1.setNote("下单");order1.setCustomer("麦迪");Order order7 = new Order();order7.setOrderNo("3");order7.setNote("下单");order7.setCustomer("库里");Order order2 = new Order();order2.setOrderNo("1");order2.setNote("付款");order2.setCustomer("麦迪");Order order3 = new Order();order3.setOrderNo("1");order3.setNote("通知");order3.setCustomer("麦迪");Order order4 = new Order();order4.setOrderNo("2");order4.setNote("下单");order4.setCustomer("科比");Order order5 = new Order();order5.setOrderNo("2");order5.setNote("付款");order5.setCustomer("科比");Order order8 = new Order();order8.setOrderNo("3");order8.setNote("付款");order8.setCustomer("库里");Order order6 = new Order();order6.setOrderNo("2");order6.setNote("通知");order6.setCustomer("科比");Order order9 = new Order();order9.setOrderNo("3");order9.setNote("通知");order9.setCustomer("库里");orders.add(order1);orders.add(order7);orders.add(order2);orders.add(order3);orders.add(order4);orders.add(order5);orders.add(order8);orders.add(order6);orders.add(order9);return orders;}public static void main(String[] args) throws Exception {DefaultMQProducer producer = new DefaultMQProducer("example_group_name");producer.setNamesrvAddr("47.103.143.191:8201;47.103.143.191:8202");producer.start();List<Order> orders = buildMsg();orders.forEach(x -> {Message msg = new Message("TopicOrder", "TagOrder4", x.getOrderNo(), JSON.toJSONBytes(x));try {SendResult sendResult = producer.send(msg, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> queues, Message message, Object arg) {Integer orderNo = Integer.valueOf((String) arg);int index = orderNo % queues.size();//          log.info("order: {}, queueTotal: {}, index: {}, queue: {}", JSON.toJSONString(x), queues.size(),//                    index, queues.get(index));log.info("客户: {}, 发送到的队列: {}", x.getCustomer(), index);return queues.get(index);}}, x.getOrderNo());log.info("sendResult: {}", JSON.toJSONString(sendResult));} catch (MQClientException e) {e.printStackTrace();} catch (RemotingException e) {e.printStackTrace();} catch (MQBrokerException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}});producer.shutdown();}}

消费者

@Slf4j
public class OrderedConsumer {public static void main(String[] args) throws Exception {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.setNamesrvAddr("47.103.143.191:8201;47.103.143.191:8202");consumer.subscribe("TopicOrder", "*");consumer.registerMessageListener(new MessageListenerOrderly() {@SneakyThrows@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {context.setAutoCommit(false);log.info("线程：{}, 队列：{}, 消费消息：{}", Thread.currentThread().getName(), msgs.get(0).getQueueId(),new String(msgs.get(0).getBody(), "UTF-8"));return ConsumeOrderlyStatus.SUCCESS;}});consumer.start();System.out.printf("Consumer Started.%n");}
}

结果验证

生产者发送日志，可以看出，同一个人的业务发送到了同一个队列，总共9条消息

消费者

当生产者第一次发送9条消息时，看下图红字框中，没问题，3个线程，分别消费3个队列中的消息，但是当我再次发送消息时，消费者消费就乱了，每一个消息都由一个线程消费，不断的新建线程，4-13...再发送消息，继续创建新的线程，直到总量20，再从线程1开始消费

如果从新启动消费者，就是正常的3个线程123消费3个队列，如果不从新启动消费者，生产者不断发送，消费端就无法按顺序正常消费，而是不断创建新线程消费

疑问

1.消费者线程为什么是20个

2.消费者为什么创建新线程只消费一个，不是一个线程，消费一个队列

3.消费者多线程在哪里实现，真实的消费端多线程消费场景就是这样吗，不用管，自动会有多个线程消费，还是怎么实现多个消费线程消费

有知晓的同志请解答，谢谢

思考

队列与broker

每个broker默认有4个队列，双主双从，就是4*4 16个队列，所以稍微需要考虑一个消息均衡的发送到指定队列