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
拉取镜像
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 查看所有的已经启动的容器。
可以看到 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 = 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
第一台服务器:
第二台服务器:
大家看一下 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 能看到下面的就代表成功了!!
整合 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 地址是否正确!