SpringBoot整合RocketMQ

1、RocketMQ概念

Apache RocketMQ作为阿里开源的一款高性能、高吞吐量的分布式消息中间件。
支持Broker和Consumer端消息过滤，支持发布订阅模型和点对点，支持拉pull和推push两种消息模式，单一队列百万消息、亿级消息堆积，支持单master节点，多master节点，多master多slave节点，任意一点都是高可用，水平拓展，Producer、Consumer、队列都可以分布式，消息失败重试机制、支持特定level的定时消息，新版本底层采用Netty，4.3.x支持分布式事务，适合金融类业务，高可用性跟踪和审计功能。

Producer:消息生产者
Producer Group:消息生产者组，发送同类消息的一个消息生产组
Consumer:消费者
Consumer Group:消费同类消息的多个实例
Tag:标签，子主题（二级分类）对topic的进一步细化,用于区分同一个主题下的不同业务的消息
Topic:主题, 如订单类消息，queue是消息的物理管理单位，而topic是逻辑管理单位。一个topic下可以有多个queue，
默认自动创建是4个，手动创建是8个
Message：消息,每个message必须指定一个topic
Broker：MQ程序，接收生产的消息，提供给消费者消费的程序
Name Server：给生产和消费者提供路由信息，提供轻量级的服务发现、路由、元数据信息，可以多个部署，互相独立（比zookeeper更轻量）
Offset: 偏移量，可以理解为消息进度
commit log: 消息存储会写在Commit log文件里面

2、发送消息

1、添加maven依赖

<?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">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.ljessie</groupId>
    <artifactId>rocketmq-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rocketmq-demo</name>
    <description>Demo project for Spring Boot</description>

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

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

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.3.0</version>
        </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>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

2、新建jms包，JMSConfig类，设置配置常量

package com.ljessie.rocketmqdemo.jms;

public class JMSConfig {
    public static final String producerGroup = "pay_producer_group";
    public static final String transationProducerGroup = "transation_pay_producer_group";
    public static final String consumerGroup = "consumer_group";
    public static final String tagConsumerGroup = "tag_consumer_group";
    public static final String consumerOrdlerlyGroup = "consumer_orderly_group";
    public static final String nameServer="121.199.78.138:9876";
    public static final String TOPIC = "JESSIE_PAY_TEST_TOPIC";
    public static final String ORDERLY_TOPIC = "JESSIE_PAY_ORDERLY_TEST_TOPIC";
    public static final String producerOrderlyGroup = "pay_orderly_group";
}

3、新建生产者PayProducer

package com.ljessie.rocketmqdemo.jms;

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

@Component
public class PayProducer {

    private DefaultMQProducer producer;

    public PayProducer(){
        producer = new DefaultMQProducer(JMSConfig.producerGroup);
        //生产者投递消息重复次数
        producer.setRetryTimesWhenSendFailed(3);
        producer.setNamesrvAddr(JMSConfig.nameServer);
//        producer.setVipChannelEnabled(false);
        start();
    }

    public DefaultMQProducer getProducer(){
        return this.producer;
    }

    /**
     * 对象在使用之前，必须要调用一次，只能初始化一次
     */
    public void start(){
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutDown(){
        this.producer.shutdown();
    }
}

4、新建PayController

package com.ljessie.rocketmqdemo.controller;

import com.ljessie.rocketmqdemo.domain.ProductOrder;
import com.ljessie.rocketmqdemo.jms.JMSConfig;
import com.ljessie.rocketmqdemo.jms.PayProducer;
import com.ljessie.rocketmqdemo.jms.TransationProducer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;

@RestController
@RequestMapping("/api/")
public class PayController {

    @Autowired
    private PayProducer payProducer;

    /**
     * 同步发送消息
     * @param text
     * @return
     * @throws InterruptedException
     * @throws RemotingException
     * @throws MQClientException
     * @throws MQBrokerException
     */
    @RequestMapping("sync")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(JMSConfig.TOPIC,"taga",("Hello RocketMQ:"+text).getBytes());
        SendResult sendResult = payProducer.getProducer().send(message);
        return sendResult;
    }
}

这个时候，启动应用，访问api/sync路径，会报没有这个topic异常

org.apache.rocketmq.client.exception.MQClientException: No route info of this topic, JESSIE_PAY_TEST_TOPIC

需要去管控台手工创建topic，或者将SpringBoot依赖的RocketMQ版本与服务端RocketMQ的版本改成一样的，也可以。例如我的pom.xml文件中，依赖的版本是4.3.0，而服务器上部署的RocketMQ版本是4.4.0，就不会自动创建topic。
创建topic之后，再次访问访问api/sync路径，会报下面异常

org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout

在conf/broker.conf文件中添加如下配置

brokerIP1 = 替换成自己的公网IP

然后指定配置文件启动Broker，返回上一级，然后重启Broker

[root@iZbp1fk51wcf50rx5ngu9qZ conf]# vim broker.conf 
[root@iZbp1fk51wcf50rx5ngu9qZ conf]# cd ..
[root@iZbp1fk51wcf50rx5ngu9qZ apache-rocketmq]# jps
27058 NamesrvStartup
29515 BrokerStartup
29581 Jps
[root@iZbp1fk51wcf50rx5ngu9qZ apache-rocketmq]# kill -9 29515
[root@iZbp1fk51wcf50rx5ngu9qZ apache-rocketmq]# jps
27058 NamesrvStartup
29591 Jps
[4]+  Exit 137                nohup sh bin/mqbroker -n localhost:9876
[root@iZbp1fk51wcf50rx5ngu9qZ apache-rocketmq]# sh bin/mqbroker -n localhost:9876  -c ./conf/broker.conf
The broker[broker-a, 121.199.78.138:10911] boot success. serializeType=JSON and name server is localhost:9876

或者使用守护进程方式启动Broker

nohup sh bin/mqbroker -n localhost:9876  -c ./conf/broker.conf &

再次访问api/sync，返回成功

{"sendStatus":"SEND_OK","msgId":"C0A80A671B1018B4AAC259C57DD20001","messageQueue":{"topic":"JESSIE_PAY_TEST_TOPIC","brokerName":"broker-a","queueId":1},"queueOffset":0,"transactionId":null,"offsetMsgId":"79C74E8A00002A9F0000000000057D64","regionId":"DefaultRegion","traceOn":true}

RocketMQ管控台也可查询此消息

msgTest.jpg

3、消费消息

新建PayConsumer

package com.ljessie.rocketmqdemo.jms;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
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.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Component
public class PayConsumer {

    private DefaultMQPushConsumer consumer;

    public PayConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(JMSConfig.consumerGroup);
        consumer.setNamesrvAddr(JMSConfig.nameServer);
        //从队首开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //订阅主题，第二个参数*代表注册这个主题下面的所有tag
        consumer.subscribe(JMSConfig.TOPIC,"*");
        //配置消费监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt message = list.get(0);
                int retryTimes = message.getReconsumeTimes();
                try {
                    System.out.println("Consumer消费成功:"+new String(message.getBody(),"utf-8"));
                } catch (UnsupportedEncodingException e) {
                    //如果重复次数超过两次不成功，则人工介入。广播方式不提供重试机制。
                    if(retryTimes >= 2){
                        //告诉Broker消费成功
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    e.printStackTrace();
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}

访问：http://localhost:8080/api/sync?text=test
控制台打印：

Consumer消费成功:Hello RocketMQ:test