Springboot2.x整合RocketMQ4.x实战发送消息 - 超级大汇总——DBC的博客超级大汇总—

加入相关依赖

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

温馨提示

自动创建topic： autoCreateTopicEnable=true 无效原因：客户端版本要和服务端版本保持一致

老规矩，看一下结构图

PayProducer

package net.xdclass.xdclassmq.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 String producerGroup = "pay_group";

    private String nameServerAddr = "81.71.147.62:9876";


    private DefaultMQProducer producer;


    public  PayProducer(){
        producer = new DefaultMQProducer(producerGroup);

        //指定NameServer地址，多个地址以 ; 隔开
        //如 producer.setNamesrvAddr("192.168.100.141:9876;192.168.100.142:9876;192.168.100.149:9876");

        producer.setNamesrvAddr(nameServerAddr);
        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();
    }


}

PayController

package net.xdclass.xdclassmq.controller;

import net.xdclass.xdclassmq.jms.PayProducer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
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.HashMap;

@RestController
public class PayController {

    @Autowired
    private PayProducer payProducer;

    private static  final String topic = "xdclass_pay_test_topic";

    @RequestMapping("/api/v1/pay_cb")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {

        Message message = new Message(topic,"taga", ("hello xdclass rocketmq = "+text).getBytes() );

        SendResult sendResult = payProducer.getProducer().send(message);
        System.out.println(sendResult);

        return new HashMap<>();
    }

}

直接启动就好了，成功如下图

Message对象
- topic: 主题名称
- tag: 标签，用于过滤
- key: 消息唯一标示，可以是业务字段组合
- body: 消息体,字节数组
注意发送消息到Broker，需要判断是否有此topic启动broker的时候，
本地环境建议开启自动创建topic，生产环境建议关闭自动化创建topic

建议先手工创建Topic，如果靠程序自动创建，然后再投递消息，会出现延迟情况
概念模型: 一个topic下面对应多个queue,可以在创建Topic时指定，如订单类topic
通过可视化管理后台查看消息

常见错误一

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

原因：阿里云存在多网卡，rocketmq都会根据当前网卡选择一个IP使用，当你的机器有多块网卡时，很有可能会有问题。比如，我遇到的问题是我机器上有两个IP，一个公网IP，一个私网IP, 因此需要配置broker.conf 指定当前的公网ip, 然后重新启动broker
新增配置：conf/broker.conf (属性名称brokerIP1=broker所在的公网ip地址 )
新增这个配置：brokerIP1=120.76.62.13

启动命令：nohup sh bin/mqbroker -n localhost:9876 -c ./conf/broker.conf &

如图所示：

常见错误二

MQClientException: No route info of this topic, TopicTest1
原因：Broker 禁止自动创建 Topic，且用户没有通过手工方式创建此Topic, 或者broker和Nameserver网络不通
解决：
通过 sh bin/mqbroker -m 查看配置
autoCreateTopicEnable=true 则自动创建topic

Centos7关闭防火墙 systemctl stop firewalld

常见错误三

控制台查看不了数据，提示连接 10909错误

原因：Rocket默认开启了VIP通道，VIP通道端口为10911-2=10909

解决：阿里云安全组需要增加一个端口 10909

其他错误:

https://blog.csdn.net/qq_14853889/article/details/81053145
https://blog.csdn.net/wangmx1993328/article/details/81588217#%E5%BC%82%E5%B8%B8%E8%AF%B4%E6%98%8E
https://www.jianshu.com/p/bfd6d849f156
https://blog.csdn.net/wangmx1993328/article/details/81588217

本文作者为DBC，转载请注明。