Spring Cloud Alibaba 学习 -- 4、整合RocketMQ（安装、生产、消费）

川长思鸟来 2022-10-10 09:57 8阅读 0赞

> 学习视频（B站）：https://www.bilibili.com/video/BV1Mt4y1i7JW  
> GitHub 源码地址：https://github.com/tyronczt/spring-cloud-alibaba-learning  
> 系列文章：https://blog.csdn.net/tian330726/article/details/112910508

基本概念：RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息，Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应一台服务器，每个 Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。Message Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。ConsumerGroup 由多个Consumer 实例构成。

详见：https://github.com/apache/rocketmq/blob/master/docs/cn/concept.md

### 一、Linux 安装 ###

#### 1、下载压缩包，rocketmq-all-4.9.0-bin-release.zip ####

网址：http://rocketmq.apache.org/dowloading/releases/

#### 2、解压缩 ####

unzip rocketmq-all-4.9.0-bin-release.zip

#### 3、启动 NameServer ####

cd bin
    nohup ./mqnamesrv &

#### 4、检查是否启动成功 ####

查看端口启动情况：netstat -an | grep 9876
    查看启动日志：tail -f ~/logs/rocketmqlogs/namesrv.log

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center]

#### 5、启动 Broker ####

启动之前需要编辑配置文件，修改 JVM 内存设置，默认给的内存 4 GB，超过现有虚拟机 JVM ，不然会报错  
![在这里插入图片描述][20210627170551139.png_pic_center]

cd bin
    vim runserver.sh
    vim runbroker.sh
    -Xms256m -Xmx256m -Xmn128m
    Xms为虚拟机初始化堆大小，Xmx为最大堆大小，Xmn为最小堆大小

![在这里插入图片描述][20210627170618531.png_pic_center]

启动 Broker

nohup ./mqbroker -n localhost:9876 &

查看启动情况

tail -f ~/logs/rocketmqlogs/broker.log

![在这里插入图片描述][20210627170633948.png_pic_center]

#### 6、测试 RocketMQ ####

测试消息发送

cd bin
    export NAMESRV_ADDR=localhost:9876 【参考官网：RocketMQ provides multiple ways to achieve this. For simplicity, we use environment variable NAMESRV_ADDR，否则会报错：connect to null failed】
    ./tools.sh org.apache.rocketmq.example.quickstart.Producer

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 1]

测试消息接收

cd bin
    export NAMESRV_ADDR=localhost:9876
    ./tools.sh org.apache.rocketmq.example.quickstart.Consumer

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 2]

#### 7、关闭 RocketMQ ####

cd bin
    ./mqshutdown broker
    ./mqshutdown namesrv

![在这里插入图片描述][2021062717072628.png_pic_center]

#### 8、报错 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 3]

#关闭RocketMq
    ./mqshutdown broker
    ./mqshutdown namesrv
    
    nohup ./mqnamesrv &
    
    cd ../conf/
    vim broker.conf
    
    #最后两行进行添加
    namesrvAddr=192.168.255.100:9876
    brokerIP1=192.168.255.100
    
    cd ..
    nohup sh bin/mqbroker -n 192.168.255.100:9876 autoCreateTopicEnable=true -c conf/broker.conf &

参考：https://rocketmq.apache.org/docs/quick-start/

### 二、控制台安装 ###

#### 1、下载源码，rocketmq-console ####

网址：https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console

#### 2、打包 ####

mvn clean package -Dmaven.test.skip=true

#### 3、修改配置文件 ####

properties
    server.port=9877
    rocketmq.config.namesrvAddr=192.168.255.100:9876
    rocketmq.config.isVIPChannel=false

#### 4、运行 ####

java -jar target/rocketmq-console-ng-2.0.0.jar

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 4]

http://localhost:9877/  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 5]

#### 5、报错 ####

rocketmq:connect to 172.17.0.1:10911 failed

这是因为我们的 RocketMQ 安装在 Linux 中，控制台在 windows，Linux 需要开放端口才能访问，开放 10909 和 10911和9876 端口

firewall-cmd --zone=public --add-port=10909/tcp --permanent
    firewall-cmd --zone=public --add-port=10911/tcp --permanent
    firewall-cmd --zone=public --add-port=9876/tcp --permanent
    systemctl restart firewalld.service
    firewall-cmd --reload

### 三、Java 实现消息发送 ###

在provider项目下

#### 1、引入依赖 ####

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.1.0</version>
    </dependency>

#### 2、生产消息 ####

public class Test {
        
        public static void main(String[] args) throws Exception {
        
            //创建消息生产者
            DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
            //设置NameServer
            producer.setNamesrvAddr("192.168.255.100:9876");
            //启动生产者
            producer.start();
            //构建消息对象
            Message message = new Message("myTopic", "myTag", ("Test MQ").getBytes());
            //发送消息
            SendResult result = producer.send(message, 1000);
            System.out.println(result);
            //关闭生产者
            producer.shutdown();
        }
    }

#### 3、查看消息 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 7]

#### 4、消费消息 ####

处于阻塞状态，只要有消息发送过来，就会消费信息

@Slf4j
    public class ConsumerTest {
        
        public static void main(String[] args) throws MQClientException {
        
            //创建消息消费者
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer-group");
            //设置NameServer
            consumer.setNamesrvAddr("192.168.255.100:9876");
            //指定订阅的主题和标签
            consumer.subscribe("myTopic", "*");
            //回调函数
            consumer.registerMessageListener(new MessageListenerConcurrently() {
        
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        
                    log.info("Message=>{}", list);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            //启动消费者
            consumer.start();
        }
    }

5、查看消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 8]

### 四、SpringBoot整合RocketMQ ###

#### 1、生产者引入依赖 ####

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.1.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.7.0</version>
    </dependency>

#### 2、配置文件 ####

rocketmq:
      name-server: 192.168.255.100:9876
      producer:
        group: myprovider

#### 3、生产者代码 ####

@Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    @GetMapping("/create")
    public Order create(){
        
        Order order = new Order(
            1,
            "张三",
            "123123",
            "软件园",
            new Date()
        );
        this.rocketMQTemplate.convertAndSend("myTopic",order);
        return order;
    }

#### 4、消费者引入依赖 ####

#### 5、配置文件 ####

rocketmq:
      name-server: 192.168.255.100:9876

#### 6、消费者代码 ####

@Slf4j
    @Service
    @RocketMQMessageListener(consumerGroup = "myConsumer", topic = "myTopic")
    public class SmsService implements RocketMQListener<Order> {
        
        @Override
        public void onMessage(Order order) {
        
            log.info("新订单{},发短信", order);
        }
    }

只要生产者中的topic与消费者中的topic进行对应即可。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 9]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/0a7be8593cda44c682c53d3def5f734c.png
[20210627170551139.png_pic_center]: /images/20221005/0e3e513b9caf49c6bc6892a13dca2475.png
[20210627170618531.png_pic_center]: /images/20221005/e5b6b30ea4d448a498f2e774a4becf2a.png
[20210627170633948.png_pic_center]: /images/20221005/40df0cee7d544379b1b947e3f4483cf9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221005/50472b98ad0145cd868b28286d764f79.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221005/6e2f8aa6f83d44b2bcbec257d6ec2b51.png
[2021062717072628.png_pic_center]: /images/20221005/c17de38a5fb64b75a508982d6e6f2661.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221005/0bfbf92c89a94332ae9834f1446d351d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221005/46df1779de6d4d80b0c5555581304b6f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221005/49e7e351807543c5b8d4d68646827df3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221005/522dcf6ff0784a12a5db5f41965e37af.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221005/bae63cda36c54d33bbd515b8958966ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221005/e9e90645f4b34f858e0d8cfbd549520b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW4zMzA3MjY_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221005/ccc5b875baaf4e08a357f3ee627f02ed.png