成为消息队列专业人士，我们需要的仅仅是RocketMQ4.6.0的消费知识

2023-05-28 07:36:20

集群化、推送式、并行消费：RocketMQ 普通消息消费入门指南

引言

RocketMQ 是当今流行的消息队列技术，本文将深入探讨 RocketMQ 4.6.0 普通消息消费的实现原理，让你快速上手。普通消息消费支持集群部署，即多个消费者节点同时消费同一个消息队列。在这样的场景下，消息会以推送的方式发送到消费者。此外，消费者还可以设置并行消费，即每个消费者节点同时消费多个消息队列。

创建消息队列

bin/mqadmin createTopic -n YourTopic -b 1

该命令将创建一个名为 "YourTopic" 的消息队列，其中 "b 1" 表示该队列只有一个分区。

创建消费者组

bin/mqadmin createConsumerGroup -n YourConsumerGroup

该命令将创建一个名为 "YourConsumerGroup" 的消费者组。

启动消费者

Properties properties = new Properties();
properties.put(PropertyKeyConst.ConsumerId, "YourConsumerId");
properties.put(PropertyKeyConst.NamesrvAddr, "127.0.0.1:9876");
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("YourConsumerGroup", properties);
consumer.subscribe("YourTopic", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.println("Received message: " + new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
consumer.start();

这段 Java 代码创建一个 DefaultMQPushConsumer，它是一个推送式消费者。我们将它注册为 "YourConsumerGroup" 的成员，并订阅 "YourTopic" 中所有 tag 的消息。当收到消息时，我们会打印消息内容并将其标记为已消费。

生产消息

Properties properties = new Properties();
properties.put(PropertyKeyConst.ProducerId, "YourProducerId");
properties.put(PropertyKeyConst.NamesrvAddr, "127.0.0.1:9876");
DefaultMQProducer producer = new DefaultMQProducer("YourProducerGroup", properties);
producer.start();
for (int i = 0; i < 10; i++) {
    Message msg = new Message("YourTopic", "YourTag", ("Hello RocketMQ " + i).getBytes());
    producer.send(msg);
}
producer.shutdown();