springboot 如何优雅的使用 alibaba rocketmq 消息队列-阿里云开发者社区

文章保证，一路执行下去不会报错，报错你找我。如果觉得有用，希望屏幕前的大佬，点赞➕关注

只有想到的方面多时，才能在够实际运用中发现更多的问题，这也是作者给读者最好的礼物

新建 springboot 项目

新建一个 springboot 项目，pom 文件如下，rocketmq 使用 apache 封装的 jar

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.11.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>cn.machen.study</groupId>
    <artifactId>cloud-alibaba-rocketmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <modelVersion>4.0.0</modelVersion>

    <properties>
        <java.version>1.8</java.version>
    </properties>

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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

    <build>
        <pluginManagement>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </pluginManagement>
    </build>

</project>

创建配置管理类

我们 nameserver 直接使用本地安装的就可以，topic 是 rocketmq 服务端默认就会创建的

public class RmqConfig {
    /**
     * Name Server 地址，因为是集群部署 所以有多个用 分号 隔开
     */
    public static final String NAME_SERVER = "localhost:9876";
    /**
     * 主题名称 主题一般是服务器设置好 而不能在代码里去新建topic（ 如果没有创建好，生产者往该主题发送消息 会报找不到topic错误）
     */
    public static final String TOPIC = "TestTopic";
}

Producer

新建消息生产端 Producer，负责生产消息

import lombok.Getter;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.stereotype.Component;

@Getter
@Component
public class Producer {
    /**
     * 生产者组
     */
    private static final String producerGroup = "test_producer";
    private DefaultMQProducer producer;

    public Producer() throws MQClientException {
        producerInit().start();
    }

    private DefaultMQProducer producerInit() {
        producer = new DefaultMQProducer(producerGroup);
        // 不开启 vip 通道 开通口端口会减 2
        producer.setVipChannelEnabled(false);
        // 绑定 name server
        producer.setNamesrvAddr(RmqConfig.NAME_SERVER);
        return producer;
    }
}

Consumer

建立消息消费端 Consumer，负责消费消息

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;

@Slf4j
@Component
public class Consumer {
    /**
     * 消费者组
     */
    private static final String CONSUMER_GROUP = "test_consumer";

    public Consumer() throws MQClientException {
        consumerInit().start();
    }

    private DefaultMQPushConsumer consumerInit() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
        consumer.setNamesrvAddr(RmqConfig.NAME_SERVER);
        // 消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 订阅主题和 标签（ * 代表所有标签)下信息
        consumer.subscribe(RmqConfig.TOPIC, "*");
        // 注册消费的监听 并在此监听中消费信息, 并返回消费的状态信息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            // msgs 中只收集同一个topic, 同一个 tag, 并且 key 相同的message
            // 会把不同的消息分别放置到不同的队列中
            try {
                for (Message msg : msgs) {
                    // 消费者获取消息 这里只输出 不做后面逻辑处理
                    String body = new String(msg.getBody(), "utf-8");
                    log.info("  >>> [消费端] 获取消息主题 topic :: {}, 消费消息 body :: {} ", msg.getTopic(), body);
                }
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        return consumer;
    }
}

测试 Controller

创建对应 Controller 测试类，生产 100 条消息由消费者执行，最终把返回信息进行打印

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
public class Controller {

    @Autowired
    private Producer producer;

    @RequestMapping("/rocketmq/send")
    public Object callback() throws Exception {
        for (int i = 0; i < 100; i++) {
            // 创建生产信息
            Message message = new Message(RmqConfig.TOPIC, "testTag", (String.format("[生产者] 发送第 【%d】 次 mq 消息", i)).getBytes());
            // 发送消息
            SendResult sendResult = producer.getProducer().send(message);
            log.info("  >>> [生产者] 发送消息返回对象 :: {}", sendResult);
        }

        return "执行成功";
    }
}

截取部分日志，输出打印如下：

2020-12-12 15:31:44.718  INFO 45296 --- [essageThread_11] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【90】 次 mq 消息 
2020-12-12 15:31:44.720  INFO 45296 --- [nio-8080-exec-1] c.m.s.cloudalibabarocketmq.Controller    :   >>> [生产者] 发送消息返回对象 :: SendResult [sendStatus=SEND_OK, msgId=0A10000EB0F018B4AAC23BFAFDCD005B, offsetMsgId=C0A8010200002A9F000000000027CF47, messageQueue=MessageQueue [topic=TestTopic, brokerName=broker-a, queueId=0], queueOffset=3060]
2020-12-12 15:31:44.721  INFO 45296 --- [essageThread_12] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【91】 次 mq 消息 
2020-12-12 15:31:44.722  INFO 45296 --- [nio-8080-exec-1] c.m.s.cloudalibabarocketmq.Controller    :   >>> [生产者] 发送消息返回对象 :: SendResult [sendStatus=SEND_OK, msgId=0A10000EB0F018B4AAC23BFAFDD0005C, offsetMsgId=C0A8010200002A9F000000000027D018, messageQueue=MessageQueue [topic=TestTopic, brokerName=broker-a, queueId=1], queueOffset=3062]
2020-12-12 15:31:44.724  INFO 45296 --- [essageThread_13] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【92】 次 mq 消息 
2020-12-12 15:31:44.725  INFO 45296 --- [nio-8080-exec-1] c.m.s.cloudalibabarocketmq.Controller    :   >>> [生产者] 发送消息返回对象 :: SendResult [sendStatus=SEND_OK, msgId=0A10000EB0F018B4AAC23BFAFDD3005D, offsetMsgId=C0A8010200002A9F000000000027D0E9, messageQueue=MessageQueue [topic=TestTopic, brokerName=broker-a, queueId=2], queueOffset=3059]
2020-12-12 15:31:44.726  INFO 45296 --- [essageThread_14] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【93】 次 mq 消息 
2020-12-12 15:31:44.728  INFO 45296 --- [nio-8080-exec-1] c.m.s.cloudalibabarocketmq.Controller    :   >>> [生产者] 发送消息返回对象 :: SendResult [sendStatus=SEND_OK, msgId=0A10000EB0F018B4AAC23BFAFDD5005E, offsetMsgId=C0A8010200002A9F000000000027D1BA, messageQueue=MessageQueue [topic=TestTopic, brokerName=broker-a, queueId=3], queueOffset=3058]
2020-12-12 15:31:44.729  INFO 45296 --- [essageThread_15] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【94】 次 mq 消息 
2020-12-12 15:31:44.731  INFO 45296 --- [nio-8080-exec-1] c.m.s.cloudalibabarocketmq.Controller    :   >>> [生产者] 发送消息返回对象 :: SendResult [sendStatus=SEND_OK, msgId=0A10000EB0F018B4AAC23BFAFDD8005F, offsetMsgId=C0A8010200002A9F000000000027D28B, messageQueue=MessageQueue [topic=TestTopic, brokerName=broker-a, queueId=0], queueOffset=3061]
2020-12-12 15:31:44.732  INFO 45296 --- [essageThread_16] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【95】 次 mq 消息 
2020-12-12 15:31:44.734  INFO 45296 --- [nio-8080-exec-1] c.m.s.cloudalibabarocketmq.Controller    :   >>> [生产者] 发送消息返回对象 :: SendResult [sendStatus=SEND_OK, msgId=0A10000EB0F018B4AAC23BFAFDDB0060, offsetMsgId=C0A8010200002A9F000000000027D35C, messageQueue=MessageQueue [topic=TestTopic, brokerName=broker-a, queueId=1], queueOffset=3063]
2020-12-12 15:31:44.735  INFO 45296 --- [essageThread_17] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【96】 次 mq 消息 
2020-12-12 15:31:44.737  INFO 45296 --- [nio-8080-exec-1] c.m.s.cloudalibabarocketmq.Controller    :   >>> [生产者] 发送消息返回对象 :: SendResult [sendStatus=SEND_OK, msgId=0A10000EB0F018B4AAC23BFAFDDE0061, offsetMsgId=C0A8010200002A9F000000000027D42D, messageQueue=MessageQueue [topic=TestTopic, brokerName=broker-a, queueId=2], queueOffset=3060]
2020-12-12 15:31:44.738  INFO 45296 --- [essageThread_18] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【97】 次 mq 消息 
2020-12-12 15:31:44.740  INFO 45296 --- [nio-8080-exec-1] c.m.s.cloudalibabarocketmq.Controller    :   >>> [生产者] 发送消息返回对象 :: SendResult [sendStatus=SEND_OK, msgId=0A10000EB0F018B4AAC23BFAFDE10062, offsetMsgId=C0A8010200002A9F000000000027D4FE, messageQueue=MessageQueue [topic=TestTopic, brokerName=broker-a, queueId=3], queueOffset=3059]
2020-12-12 15:31:44.741  INFO 45296 --- [essageThread_19] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【98】 次 mq 消息 
2020-12-12 15:31:44.742  INFO 45296 --- [nio-8080-exec-1] c.m.s.cloudalibabarocketmq.Controller    :   >>> [生产者] 发送消息返回对象 :: SendResult [sendStatus=SEND_OK, msgId=0A10000EB0F018B4AAC23BFAFDE40063, offsetMsgId=C0A8010200002A9F000000000027D5CF, messageQueue=MessageQueue [topic=TestTopic, brokerName=broker-a, queueId=0], queueOffset=3062]
2020-12-12 15:31:44.745  INFO 45296 --- [essageThread_20] c.m.study.cloudalibabarocketmq.Consumer  :   >>> [消费端] 获取消息主题 topic :: TestTopic, 消费消息 body :: [生产者] 发送第 【99】 次 mq 消息

根据文章这么一套走下来，可以说是玩过 rocketmq 了，后面就需要更深入去了解 rocketmq 特性以及具体使用了

springboot 如何优雅的使用 alibaba rocketmq 消息队列

新建 springboot 项目

创建配置管理类

Producer

Consumer

测试 Controller

云消息队列

热门文章

最新文章

相关产品

相关课程

相关电子书

相关实验场景