rocketmq 消费顺序消息 java实现负载均衡

在 RocketMQ 中，消费顺序消息时，可以使用两种方式来实现负载均衡：

1. 使用 MessageListenerOrderly 接口

RocketMQ 提供了 MessageListenerOrderly 接口，它可以让消费者按照消息的顺序依次消费。同时，它还提供了一个参数，可以让我们设置消息消费者的线程数。例如：

public class OrderlyMessageListener implements MessageListenerOrderly {
    private final AtomicInteger threadIndex = new AtomicInteger(0);

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        int index = threadIndex.getAndIncrement();
        // TODO: 处理消息的业务逻辑
        return ConsumeOrderlyStatus.SUCCESS;
    }
}

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
consumer.setConsumeThreadMin(1);
consumer.setConsumeThreadMax(10);
consumer.registerMessageListener(new OrderlyMessageListener());
consumer.subscribe("topic", "*");
consumer.start();

在上面的代码中，我们通过 setConsumeThreadMin 和 setConsumeThreadMax 方法来设置消费者的线程数。当 RocketMQ 推送消息给消费者时，它会根据线程数来进行负载均衡，保证每个线程都有消息可以处理。

2. 使用 MessageListenerConcurrently 接口

除了 MessageListenerOrderly 接口外，RocketMQ 还提供了 MessageListenerConcurrently 接口，它允许多个线程并发消费消息。同样，它也提供了一个参数，可以让我们设置消息消费者的线程数。例如：

public class ConcurrentMessageListener implements MessageListenerConcurrently {
    private final AtomicInteger threadIndex = new AtomicInteger(0);

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        int index = threadIndex.getAndIncrement();
        // TODO: 处理消息的业务逻辑
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
consumer.setConsumeThreadMin(1);
consumer.setConsumeThreadMax(10);
consumer.registerMessageListener(new ConcurrentMessageListener());
consumer.subscribe("topic", "*");
consumer.start();

在上面的代码中，我们同样通过 setConsumeThreadMin 和 setConsumeThreadMax 方法来设置消费者的线程数。当 RocketMQ 推送消息给消费者时，它会根据线程数来进行负载均衡，保证每个线程都有消息可以处理。由于使用的是并发消费，所以需要注意消息处理的顺序可能不是严格按照消息顺序的。