RocketMQ消息实战：通俗易懂，深入浅出

2024-02-24 23:12:46

作为一名在技术博客创作领域独树一帜的专家，我致力用情感丰富的文字和精准的语言，打造引人入胜的文章。在文章中，我不会直接阐释观点，而是巧妙地让这些观点引导读者思考，激发他们的共鸣。在掌握SEO技巧的前提下，我通过灵活运用关键词和长尾关键词，确保文章在搜索引擎中的可见度。

今天，我将结合之前对RocketMQ源码的研究成果，通过一系列通俗易懂的案例，带领大家深入浅出地了解RocketMQ的各种消息发送和消费方式。

RocketMQ作为一款优秀的分布式消息队列系统，广泛应用于各种场景。下面我们就来通过实战演练，体验一下RocketMQ的强大功能。

发送普通消息

// 创建生产者对象
Producer producer = new DefaultMQProducer("group1");

// 设置name server地址
producer.setNamesrvAddr("127.0.0.1:9876");

// 启动生产者
producer.start();

// 创建消息对象
Message message = new Message("topic1", "普通消息");

// 发送消息
producer.send(message);

// 关闭生产者
producer.shutdown();

消费普通消息

// 创建消费者对象
Consumer consumer = new DefaultMQPushConsumer("group1");

// 设置name server地址
consumer.setNamesrvAddr("127.0.0.1:9876");

// 设置订阅主题
consumer.subscribe("topic1", "*");

// 注册消息监听器
consumer.registerMessageListener(new MessageListenerConcurrently() {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        // 消费消息
        for (MessageExt message : messages) {
            System.out.println(new String(message.getBody()));
        }

        // 返回消费状态
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

// 启动消费者
consumer.start();