RocketMQ 入门-(整合 springboot) 单机部署 & 集群部署

springBoot+RocketMQ 示例

创建 springboot 项目或者在自己的项目中的 pom 文件中引入依赖

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

引入依赖之后我们需要配置一下 rocketmq 的配置信息，打开我们的 application.yml

rocketmq:
 name-server: 8.***.***.**:9876;  #nameserver IP地址
 producer:
   group: TEST_GROUP # 指定group
   send-message-timeout: 3000 # 消息发送超时时长，默认3s
   retry-times-when-send-failed: 3 # 同步发送消息失败重试次数，默认2
   retry-times-when-send-async-failed: 3 # 异步发送消息失败重试次数，默认2

定义生产者

首先自定义一个生产者。用来投递消息。

@Slf4j
@Component
public class MQProducerService {

   // 直接注入使用，用于发送消息到broker服务器
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

   /**
     * 普通发送
     */
    public void send(String body,String topic) {
        rocketMQTemplate.convertAndSend(topic , body);
//        rocketMQTemplate.send(topic + ":tag1", MessageBuilder.withPayload(user).build()); // 等价于上面一行
    }

    /**
     * 发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
     * （msgBody也可以是对象，sendResult为返回的发送结果）
     */
    public SendResult sendMsg(String body,String topic) {
        SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(body).build());
        log.info("【sendMsg】sendResult={}", JSON.toJSONString(sendResult));
        return sendResult;
    }

   /**
     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
     * （适合对响应时间敏感的业务场景）
     */
    public void sendAsyncMsg(String body,String topic) {

        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(body).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                //发送成功处理...
            }
            @Override
            public void onException(Throwable throwable) {
              //发送失败处理...
        
            }
        });
    }
    
   /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public void sendDelayMsg(String body,String topic, int delayLevel) {
        rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(body).build(), 3000, delayLevel);
    }

    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果，如日志）
     */
    public void sendOneWayMsg(String body,String topic) {
        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(body).build());
    }
    
   /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     */
    public SendResult sendTagMsg(String body,String topic) {
        return rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(body).build());
    }

}

代码中 body 就是消息体，如果你发消息之前是一个/一组对象，可以转换成 json 格式，当然也可以把上面的 body 修改成自己要传的对象比如 User。 topic 就是要发送到那个主题上。

 public void sendOneWayMsg(User user,String topic) {
        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(user).build());
    }

上面就是一个 Producer 示例，里面包含了多种发送消息的模式，当然 RocketMQ 给我们提供的不止这些，大家可以去官方文档看一下。

定义消费者

@Slf4j
// MessageExt：是一个消息接收通配符，不管发送的是String还是对象，都可接收，当然也可以像上面明确指定类型（我建议还是指定类型较方便）
//,messageModel = MessageModel.BROADCASTING
@Service
@RocketMQMessageListener(topic = "TEST_TOPIC", consumerGroup = "TEST_Group")
public class TaxInfoConsumerService implements RocketMQListener<MessageExt> {
   
    @Override
    public void onMessage(MessageExt message) {
        System.out.println("线程" + Thread.currentThread() + "内容为:"
                + new String(message.getBody()) +
                "队列序号:" + message.getQueueId());
        //这里消费的消息可以写自己的业务逻辑代码，比如插入，删除，上传。。。

    }

这个就是简单的消费者示例。

@RocketMQMessageListener(topic = "TEST_TOPIC", consumerGroup = "Test_Group")

这行代码表示当前消费者监听了 TEST_TOPIC 上的信息，消费者组的名字就叫做 Test_Group. 如果有生产者往 TEST_TOPIC 上投递消息，就会被当前消费者感知，并且去消费指定 Topic 上的信息。

发送消息

//使用之前先注入我们刚创建的MQProducerService
@Inject
private MQProducerService mqProducerService;



......
//我自己有一个类TaxInfoBean,里面封装了我要传的数据：
TaxInfoBean taxInfoBean = new TaxInfoBean();
taxInfoBean.setSn(taxInfoSn.nextNo());
taxInfoBean.setOutTxNo(outTxNo);
taxInfoBean.setNotifyTimes(0);
taxInfoBean.setAmount(amount);
taxInfoBean.setIdCardNo(idCardNo);
taxInfoBean.setTaxNo(taxNo);
taxInfoBean.setInvoiceCode(invoiceCode);
taxInfoBean.setBankReceiptFiles(bankReceiptFiles);
taxInfoBean.setProofFiles(proofFiles);
taxInfoBean.setTaxProjectName(taxProjectName);
String taxInfoData = JSON.toJSONString(taxInfoBean);

//重点：使用刚刚创建的mqProducerService的异步发送方式发送消息。
//taxinfoData是投递的消息，TEST_TOPIC代表我要投递到这个topic中
mqProducerService.sendAsyncMsg(taxInfoData, “TEST_TOPIC”);

大家可以自己创建一个测试类，测试一下，我这里只截取了部分代码。

启动测试方法，发送消息，我们打开控制台就会看到输出了：

线 程 Thread[ConsumeMessageThread_14,5,main]内 容 为 :
{
    "amount":999.99,
    "idCardNo":"410425128710026151",
    "invoiceCode":"304990000123400000001014",
    "invoiceType":1,
    "notifyTimes":0,
    "outTxNo":"BZD44V5Dh_V_vQ7h9fEXa",
    "settlementType":0,
    "sn":"1123041202077",
    "taskStatus":"PROCESSING",
    "taxNo":"91110105MA01R54K8A2",
    "taxProjectName":"20230491110105MA01R54K8A2"
}
队 列 序 号 :8

没错，这就是消费者已经成功消费了，并把整个 message 打印了出来，我们的消息内容是 message.getBody()的所有内容。

我们可以在消费者里面加上自己业务代码。

springboot+rocketmq 单机部署（centos7+docker）

想了一下还是跟大家一起用 docker 先单机部署一下 熟悉流程之后，后面的集群部署也就非常简单了。

首先打开我们的 Linux 环境，启动 docker（没有安装 docker 的可以搜一下，直接安装就行。）

启动 doker: systemctl start docker

搜索镜像

docker search rocketmq

b2338ab32ac71550882fefe9a424ae5d.png

拉取镜像

docker pull rocketmqinc/rocketmq

一般情况下拉取的就是最新版本。

这个时候 rocketmq 已经准备好了，我们回忆下上一节的启动过程，这里有些类似的地方，我们创建一个存放 nameserver 的数据目录：

mkdir -p /docker/rocketmq/nameserver/logs /docker/rocketmq/nameserver/store

同理，我们也提前创建存放 broker 配置信息目录，我们需要单独创建一个存放 broker 配置信息的文件目录

mkdir -p /docker/rocketmq/data/broker-master/confmkdir -p /docker/rocketmq/data/broker-slave/conf

能看出来，这两个文件夹就是代表 broker 的一主一从。除了这个 conf 文件夹，我们还要另外创建两个文件夹

一主一从： 前面大家都知道了 broker 可以配置主从节点，一个 master 一个 slave，master 节点会进行异步/同步刷盘到 slave 上，这也保证了如果某个 broker 挂了，slave 能够迅速顶上。

RocketMQ 支持两种刷盘方式：同步刷盘和异步刷盘。同步刷盘是指在消息发送的同时将消息写入磁盘，确保消息不会丢失，但是会降低消息发送的速度。异步刷盘是指将消息先写入内存缓存，然后再定时或者触发条件下将缓存中的消息写入磁盘，这样可以提高消息发送的速度，但是可能会造成消息丢失的风险。

mkdir -p /docker/rocketmq/data/broker-master/store 
mkdir -p /docker/rocketmq/data/broker-master/logs  

mkdir -p /docker/rocketmq/data/broker-slave/store 
mkdir -p /docker/rocketmq/data/broker-slave/logs

整体的架构就是:

?  docker/rockermq
.
├── data
│   ├── broker-master
│       ├── conf
│       ├── store
│       └── logs
│   ├── beoker-slave
│       ├── conf
│       ├── store
│       └── logs
└── nameserver
    ├── store
    └── logs

初始的文件夹已经创建好了，我们要先启动 nameserver.

docker run -d --restart=always --name rmqnamesrv --privileged=true -p 9876:9876  -v /docker/rocketmq/nameserver/logs:/root/logs -v /docker/rocketmq/nameserver/store:/root/store -e "MAX_POSSIBLE_HEAP=100000000" rocketmqinc/rocketmq sh mqnamesrv

参数说明
[图片上传失败...(image-723cc7-1688441580784)]

配置 broker

我们有关 broker 的文件夹之前已经创建好了，但是缺少配置文件 broker.conf

创建 master 的 broker.conf

brokerName = brokera-master
brokerId = 0   #id为0代表当前是主节点（master）
deleteWhen = 04
fileReservedTime = 48
namesrvAddr = 8.*.*.*:9876;  #你的nameserver地址
brokerRole = ASYNC_MASTER  
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false  #这个强烈建议设置成false
brokerIP1 = 当前服务器ip地址

然后把这个文件放到 data/broker-master/conf 里面

创建 slave 的 broker.conf

brokerName = brokera-slave
brokerId = 1
deleteWhen = 04
fileReservedTime = 48
namesrvAddr = 8.*.*.*:9876;
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
brokerIP1 = 当前服务器ip地址

然后把这个文件放到 data/broker-slave/conf 里面

autoCreateTopicEnable 代表是否开启自动创建 topic,这里大家设置成 false 就行，如果只是自己测试一下那就无所谓，发不到线上的话为了避免创建无用的 topic,浪费资源，建议还是关掉。

整体目录架构

?  docker/rockermq
.
├── data
│   ├── broker-master
│       ├── conf
│           ├──broker.conf
│       ├── store
│       └── logs
│   ├── beoker-slave
│       ├── conf
│           ├──broker.conf
│       ├── store
│       └── logs
└── nameserver
    ├── store
    └── logs

启动 broker

这里大家一定要谨慎，这里最容易出错，大家启动的时候一定提前看看自己的命令是否正确 我们要启动两次 broker(一主一从) 启动 brokera-master

docker run -d --restart=always --name brokera-master --link rmqnamesrv:namesrv -p 10911:10911 -p 10909:10909 --privileged=true -v /docker/rocketmq/data/broker-master/logs:/root/logs -v /docker/rocketmq/data/broker-master/store:/root/store -v /docker/rocketmq/data/broker-master/conf/broker.conf:/opt/docker/rocketmq/data/broker-master/conf/broker.conf -e "NAMESRV_ADDR=8.***:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/docker/rocketmq/data/broker-master/conf/broker.conf

大家把 NAMESRV_ADDR=8.*:9876",改成自己的 nameserver 地址。

如果前面的所有步骤都跟我的一样，正常情况下是没问题的，如果前面创建文件夹那里并没有跟我的一模一样，那你就要把命令改一下，文件目录做一下修改才行。

完事之后输入

docker ps -a 查看所有的已经启动的容器。

e79f5e4a990c19f9ecf3fbee07f877e7.png

可以看到 STATUS 那里是启动成功的 那么 master 节点的 broker 就已经启动成功了，接下来启动 slave

docker run -d --restart=always --name brokera-slave --link rmqnamesrv:namesrv -p 10124:10911 -p 10623:10909 --privileged=true -v /docker/rocketmq/data/broker-slave/logs:/root/logs -v /docker/rocketmq/data/broker-slave/store:/root/store -v /docker/rocketmq/data/broker-slave/conf/broker.conf:/opt/docker/rocketmq/data/broker-slave/conf/broker.conf -e "NAMESRV_ADDR=8.*:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/docker/rocketmq/data/broker-slave/conf/broker.conf

把 NAMESRV_ADDR 的地址换成自己的 nameserver 地址。

ea9cb6e2b241f301bc7f9782ded9abac.png

可以看到一主一从都已经启动成功了

安装可视化控制台

启动完 broker 和 nameserver 之后，我们启动可视化控制台。

拉取镜像

docker pull pangliang/rocketmq-console-ng

直接启动控制台 启动之前大家把 nameserv.addr 的地址改成自己的 nameserver 的 IP 地址

docker run -d --restart=always --name rmqadmin -e "JAVA_OPTS=-Drocketmq.namesrv.addr=122.*.*.*:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -p 8080:8080 pangliang/rocketmq-console-ng

91119b87834c5e1159119d2274a24db5.png

完事之后直接打开服务器 ip:8080 直接启动就能看到熟悉的页面

88244125d6b487ff268e63dd27018d0d.png

可以看到一主一从，我的名字是 brokera-master-a,brokera-slave-a

如果没有这两条 broker 记录大家一定要检查自己的 nameserver 地址有没有配置，着重检查 broker.conf,和启动容器命令，容器命令有多个文件目录，一定要跟自己创建的对的上。

springboot 使用

整合 Springboot 和上面的过程一模一样 ,application.yml 的 namesrv-addr 可能要改成自己 Linux 配置之后的地址， 其他代码不用改。

springboot+rocketmq 集群部署（centos7+docker）

趁热打铁，上面完成了单机部署，我这里有两台服务器，我采用的是 2 个 nameserver,broker 是 2 主 2 从。 每台服务器上是 1 个 nameserver 和 broker(一主一从)

我们按照上面的步骤每台服务器上都重复创建好存放 nameserver 和 broker 的文件目录，两台服务器都是如此

?  docker/rockermq
.
├── data
│   ├── broker-master
│       ├── conf
│           ├──broker.conf
│       ├── store
│       └── logs
│   ├── beoker-slave
│       ├── conf
│           ├──broker.conf
│       ├── store
│       └── logs
└── nameserver
    ├── store
    └── logs

两台服务器都拉取 rocketmq 镜像，然后依次启动容器（两台服务器命令一样）

docker run -d --restart=always --name rmqnamesrv --privileged=true -p 9876:9876 -v /docker/rocketmq/nameserver/logs:/root/logs -v /docker/rocketmq/nameserver/store:/root/store -e "MAX_POSSIBLE_HEAP=100000000" rocketmqinc/rocketmq sh mqnamesrv

这样就启动了 2 台 nameserver

接下来配置 broker.conf

第一台服务器上的一主一从 broker.conf

brokerClusterName = XgshDefaultCluster
brokerName = broker-master-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
namesrvAddr = 8.*.*.*:9876;8.*.*.*:9876 #大家可以看到这里有两个nameserver IP地址，用分号隔开。
brokerRole = ASYNC_MASTER  
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false  #关闭自动创建Topic
brokerIP1 = 本机服务器IP

brokerClusterName = XgshDefaultCluster
brokerName = broker-slave-a
brokerId = 1 
deleteWhen = 04
fileReservedTime = 48
namesrvAddr = 8.*.*.*:9876;8.*.*.*:9876
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
brokerIP1 = 本机服务器IP

第二台服务器上的一主一从 broker.conf

brokerClusterName = XgshDefaultCluster
brokerName = broker-master-b
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
namesrvAddr = 8.*.*.*:9876;8.*.*.*:9876 #大家可以看到这里有两个nameserver IP地址，用分号隔开。
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
brokerIP1 = 本机服务器IP


brokerClusterName = XgshDefaultCluster
brokerName = broker-slave-b
brokerId = 1
deleteWhen = 04
fileReservedTime = 48
namesrvAddr = 8.*.*.*:9876;8.*.*.*:9876 #大家可以看到这里有两个nameserver IP地址，用分号隔开。
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
brokerIP1 = 本机服务器IP

参数说明
[图片上传失败...(image-f0dbc-1688441580784)]

启动 broker

第一台服务器（命令中的 NAMESRV_ADDR 改成自己服务器启动的 2 个，中间用分号隔开）

docker run -d --restart=always --name broker-master01 --link rmqnamesrv:namesrv -p 10911:10911 -p 10909:10909 --privileged=true -v /docker/rocketmq/data/broker-master/logs:/root/logs -v /docker/rocketmq/data/broker-master/store:/root/store -v /docker/rocketmq/data/broker-master/conf/broker.conf:/opt/docker/rocketmq/data/broker-master/conf/broker.conf -e "NAMESRV_ADDR=8.*.*.*:9876;8.*.*.*:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/docker/rocketmq/data/broker-master/conf/broker.conf

docker run -d --restart=always --name broker-slave01 --link rmqnamesrv:namesrv -p 10124:10911 -p 10623:10909 --privileged=true -v /docker/rocketmq/data/broker-slave/logs:/root/logs -v /docker/rocketmq/data/broker-slave/store:/root/store -v /docker/rocketmq/data/broker-slave/conf/broker.conf:/opt/docker/rocketmq/data/broker-slave/conf/broker.conf -e "NAMESRV_ADDR=8.*.*.*:9876;8.*.*.*:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/docker/rocketmq/data/broker-slave/conf/broker.conf

第二台服务器（命令中的 NAMESRV_ADDR 改成自己服务器启动的 2 个，中间用分号隔开）

docker run -d --restart=always --name broker-master02 --link rmqnamesrv:namesrv -p 10911:10911 -p 10909:10909 --privileged=true -v /docker/rocketmq/data/broker-master/logs:/root/logs -v /docker/rocketmq/data/broker-master/store:/root/store -v /docker/rocketmq/data/broker-master/conf/broker.conf:/opt/docker/rocketmq/data/broker-master/conf/broker.conf -e "NAMESRV_ADDR=8.*.*.*:9876;8.*.*.*:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/docker/rocketmq/data/broker-master/conf/broker.conf 

docker run -d --restart=always --name broker-slave02 --link rmqnamesrv:namesrv -p 10124:10911 -p 10623:10909 --privileged=true -v /docker/rocketmq/data/broker-slave/logs:/root/logs -v /docker/rocketmq/data/broker-slave/store:/root/store -v /docker/rocketmq/data/broker-slave/conf/broker.conf:/opt/docker/rocketmq/data/broker-slave/conf/broker.conf -e "NAMESRV_ADDR=8.*.*.*:9876;8.*.*.*:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/docker/rocketmq/data/broker-slave/conf/broker.conf

broker 启动成功，我们通过命令看一下(2 台服务器都看一下)

docker ps -a

第一台服务器：

cd0c7a92394f48fea8aef826cfd22c1f.png

第二台服务器：

1cc4428af97ec8941403501db17fe221.png

大家看一下 STATUS 是都启动成功。

启动控制台

在其中一台服务器拉取可视化界面镜像(只用在其中一台启动就可以了)

docker run -d --restart=always --name rmqadmin -e "JAVA_OPTS=-Drocketmq.namesrv.addr=122.*.*.*:9876;122.*.*.*:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -p 8080:8080 pangliang/rocketmq-console-ng

访问服务器 IP:8080 能看到下面的就代表成功了！！

ba98e38fbe0fec96cd75246638db62b4.png

整合 springboot

我们只需要在 applicatiom.yml 文件中修改为 2 个 nameserver 地址用分号隔开

rocketmq:
  name-server: 8.*.*.*:9876;8.*.*.*:9876 # 两个nameserver访问地址
  producer:
    group: Pro_Group # 必须指定group
    send-message-timeout: 3000 # 消息发送超时时长，默认3s
    retry-times-when-send-failed: 3 # 同步发送消息失败重试次数，默认2
    retry-times-when-send-async-failed: 3 # 异步发送消息失败重试次数，默认2

启动项目，我们根据自己的需要去创建 Topic,然后通过开篇的示例去发送消息到消费者消费消息过程是不变的。

问题：

1.可视化界面一直加载不出来
2.因为我用的是阿里云服务器，所以访问不了大概率是 8080 这个端口号没有加安全组，大家进入到阿里云找到服务器，添加安全组，端口号 8080，建议大家设置一下授权对象，然后把自己本机 ip 输入进去，返回浏览器刷新。就能访问了。
3.老问题，（2 主 2 从）broker 不显示
4.首先我们使用 docker ps -a 查看所有启动的容器，大家可以看下自己启动的 broker,看一下 STATUS 如果不是 Up 6 hours 这种的，而是很长一串，那就是 broker 启动失败了，还是一定要看启动命令和 broker.conf 每一行配置是否正确,nameserver 地址是否正确！