RocketMQ保证顺序消费demo

2021-07-10 本文已影响0人迦叶_金色的人生_荣耀而又辉煌

上一篇 <<<SpringBoot整合RocketMQ示例
下一篇 >>>RocketMQ如何动态扩容和缩容

1.解决顺序消费的核心思路

a、相同的业务逻辑一定要放到同一个队列中
b、每个队列必须要对应同一个消费者消费
c、RocketMQ每个队列1：1均等匹配消费者

2.代码示例

2.1 jar包引入

 <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.5.1</version>
    </dependency>

2.2 相同行为的数据投放到相同的队列中—默认是使用轮询策略的

Long orderId = System.currentTimeMillis();
String insertSql = getSqlMsg("insert", orderId);
String updateSql = getSqlMsg("update", orderId);
String deleteSql = getSqlMsg("delete", orderId);
Message insertMsg = new Message(T.TOPIC, insertSql.getBytes());
Message updateMsg = new Message(T.TOPIC, updateSql.getBytes());
Message deleteMsg = new Message(T.TOPIC, deleteSql.getBytes());
producer = rocketMQTemplate.getProducer();
producer.send(insertMsg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        // 所有的消息使用同一个订单号，存放到同一个队列中。
        Long orderId = (Long) arg;
        long index = orderId % mqs.size();
        return mqs.get((int) index);
    }
}, orderId);
producer.send(updateMsg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        // 所有的消息使用同一个订单号，存放到同一个队列中
        Long orderId = (Long) arg;
        long index = orderId % mqs.size();
        return mqs.get((int) index);
    }
}, orderId);
producer.send(deleteMsg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        // 所有的消息使用同一个订单号，存放到同一个队列中
        Long orderId = (Long) arg;
        long index = orderId % mqs.size();
        return mqs.get((int) index);
    }
}, orderId);

在Springboot中可以这样写:
@Service
@RocketMQMessageListener(topic = "myTest", consumerGroup = "myTest",consumeMode = ConsumeMode.ORDERLY,consumeThreadMax = 1)
public class OrdeConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String o) {
        System.out.println("o:" + JSON.toJSONString(o));
    }
}

ConsumeMode 支持并发和顺序两种形式
consumeThreadMax 配置为1，则只能使用同一个线程消费
默认值是20，可以从DefaultMqPushConsumer类中查找到
建议配置成和队列数相同，不过打印出来的日志会显示线程不是同一个，这是线程复用机制，底层也是保证顺序消费的

2.3 消费者同一线程读取同一个队列

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
consumer.setNamesrvAddr(T.NAMESERVER);
consumer.setConsumerGroup(T.consumerGroup);
consumer.subscribe(T.TOPIC, "");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        //设置为true，则自动应答,不会重复获取
        //设置为false，则不会自动应答，程序会重复读取
        context.setAutoCommit(true);
        MessageQueue queue = context.getMessageQueue();
        for (MessageExt ext : msgs) {
            try {
                Thread.sleep((new Random()).nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName()+" 消费者获取消息 brokerName: "+queue.getBrokerName()+" topic:"+queue.getTopic()+" queueId:"+queue.getQueueId()+",队列id：" + ext.getQueueId() + ", 消息" +
                    new String(ext.getBody()) + "msgid:" + ext.getMsgId());
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});
consumer.start();

2.4 执行结果

ConsumeMessageThread_1 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:1,队列id：1, 消息{"orderId":1577796057489,"type":"insert"}msgid:0A2581022BE518B4AAC29EEF11E70000
ConsumeMessageThread_1 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:1,队列id：1, 消息{"orderId":1577796057489,"type":"update"}msgid:0A2581022BE518B4AAC29EEF11F40001
ConsumeMessageThread_1 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:1,队列id：1, 消息{"orderId":1577796057489,"type":"delete"}msgid:0A2581022BE518B4AAC29EEF11F70002
ConsumeMessageThread_2 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:3,队列id：3, 消息{"orderId":1577796132835,"type":"insert"}msgid:0A2581022BE818B4AAC29EF0382B0000
ConsumeMessageThread_2 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:3,队列id：3, 消息{"orderId":1577796132835,"type":"update"}msgid:0A2581022BE818B4AAC29EF038360001
ConsumeMessageThread_2 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:3,队列id：3, 消息{"orderId":1577796132835,"type":"delete"}msgid:0A2581022BE818B4AAC29EF038390002
ConsumeMessageThread_3 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:0,队列id：0, 消息{"orderId":1577796180976,"type":"insert"}msgid:0A2581022BEF18B4AAC29EF0F43E0000
ConsumeMessageThread_3 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:0,队列id：0, 消息{"orderId":1577796180976,"type":"update"}msgid:0A2581022BEF18B4AAC29EF0F44A0001
ConsumeMessageThread_3 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:0,队列id：0, 消息{"orderId":1577796180976,"type":"delete"}msgid:0A2581022BEF18B4AAC29EF0F44D0002

RocketMQ保证顺序消费demo

1.解决顺序消费的核心思路

2.代码示例

2.1 jar包引入

2.2 相同行为的数据投放到相同的队列中—默认是使用轮询策略的

2.3 消费者同一线程读取同一个队列

2.4 执行结果

猜你喜欢

热点阅读