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@Componentpublic 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@Injectprivate 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
拉取镜像
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
复制代码
配置 broker
我们有关 broker 的文件夹之前已经创建好了，但是缺少配置文件 broker.conf
创建 master 的 broker.conf
brokerName = brokera-masterbrokerId = 0 #id为0代表当前是主节点（master）deleteWhen = 04fileReservedTime = 48namesrvAddr = 8.*.*.*:9876; #你的nameserver地址brokerRole = ASYNC_MASTER flushDiskType = ASYNC_FLUSHautoCreateTopicEnable = false #这个强烈建议设置成falsebrokerIP1 = 当前服务器ip地址
复制代码
然后把这个文件放到 data/broker-master/conf 里面
创建 slave 的 broker.conf
brokerName = brokera-slavebrokerId = 1deleteWhen = 04fileReservedTime = 48namesrvAddr = 8.*.*.*:9876;brokerRole = SLAVEflushDiskType = ASYNC_FLUSHautoCreateTopicEnable = falsebrokerIP1 = 当前服务器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 查看所有的已经启动的容器。
可以看到 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 地址。
可以看到一主一从都已经启动成功了
安装可视化控制台
启动完 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
复制代码
完事之后直接打开服务器 ip:8080 直接启动就能看到熟悉的页面
可以看到一主一从，我的名字是 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 = XgshDefaultClusterbrokerName = broker-master-abrokerId = 0deleteWhen = 04fileReservedTime = 48namesrvAddr = 8.*.*.*:9876;8.*.*.*:9876 #大家可以看到这里有两个nameserver IP地址，用分号隔开。brokerRole = ASYNC_MASTER flushDiskType = ASYNC_FLUSHautoCreateTopicEnable = false #关闭自动创建TopicbrokerIP1 = 本机服务器IP
复制代码
brokerClusterName = XgshDefaultClusterbrokerName = broker-slave-abrokerId = 1 deleteWhen = 04fileReservedTime = 48namesrvAddr = 8.*.*.*:9876;8.*.*.*:9876brokerRole = SLAVEflushDiskType = ASYNC_FLUSHautoCreateTopicEnable = falsebrokerIP1 = 本机服务器IP
复制代码
第二台服务器上的一主一从 broker.conf
brokerClusterName = XgshDefaultClusterbrokerName = broker-master-bbrokerId = 0deleteWhen = 04fileReservedTime = 48namesrvAddr = 8.*.*.*:9876;8.*.*.*:9876 #大家可以看到这里有两个nameserver IP地址，用分号隔开。brokerRole = ASYNC_MASTERflushDiskType = ASYNC_FLUSHautoCreateTopicEnable = falsebrokerIP1 = 本机服务器IP
brokerClusterName = XgshDefaultClusterbrokerName = broker-slave-bbrokerId = 1deleteWhen = 04fileReservedTime = 48namesrvAddr = 8.*.*.*:9876;8.*.*.*:9876 #大家可以看到这里有两个nameserver IP地址，用分号隔开。brokerRole = SLAVEflushDiskType = ASYNC_FLUSHautoCreateTopicEnable = falsebrokerIP1 = 本机服务器IP
复制代码
启动 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
第一台服务器：
第二台服务器：
大家看一下 STATUS 是都启动成功。
启动控制台
在其中一台服务器拉取可视化界面镜像(只用在其中一台启动就可以了)
docker pull pangliang/rocketmq-console-ng
启动可视化控制台的命令。(namesrv.addr 后面的 ip 地址设置为两个。)
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 能看到下面的就代表成功了！！
整合 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,然后通过开篇的示例去发送消息到消费者消费消息过程是不变的。
问题：
可视化界面一直加载不出来
因为我用的是阿里云服务器，所以访问不了大概率是 8080 这个端口号没有加安全组，大家进入到阿里云找到服务器，添加安全组，端口号 8080，建议大家设置一下授权对象，然后把自己本机 ip 输入进去，返回浏览器刷新。就能访问了。
老问题，（2 主 2 从）broker 不显示
首先我们使用 docker ps -a 查看所有启动的容器，大家可以看下自己启动的 broker,看一下 STATUS 如果不是 Up 6 hours 这种的，而是很长一串，那就是 broker 启动失败了，还是一定要看启动命令和 broker.conf 每一行配置是否正确,nameserver 地址是否正确！
结语
后面我会专门把所有的问题总结一下，发一篇文章。本节篇幅很长，感谢您的耐心阅读，入门的课程到这里就结束了，后续我会跟大家一起针对优化进行更多的配置写成文章一起学习
评论