AI 助理
备案控制台
开发者社区 云原生 中间件 文章 正文

消息中间件-RocketMq

2021-09-07 320
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
注册配置 MSE Nacos/ZooKeeper，118元/月
性能测试 PTS，5000VUM额度
云原生网关 MSE Higress，422元/月
简介： RocketMq 消息队列

mq messageQuence消息队列


作用:解耦 流量削峰 处理数据


问题:


一致性


可用性


复杂性


2.mq产品比较

特性

avtiveMQ

RabbitMQ

RocketMQ

kafka

开发语音

java

erLang

java

scala

单机吞吐量

万级

万级

10w

10w

时效性

ms

us

ms

ms

可用性

高(主从架构)

高(主从架构)

非常高(分布式)

非常高(分布式)

功能特性

比较成熟,文档多

性能好

扩展性好

大数据应用


3.快速入门


3.1 安装


下载源码包或者二进制包


源码包需要手动编译


二进制包可以直接解压使用


3.2 常用命令

#启动namesrvnohup sh mqnamesrv &
#查看启动日志tail -f ~/logs/rocketmqlogs/namesrv.log
#启动brokernohup sh mqbroker &
#查看启动日志tail -f ~/logs/rocketmqlogs/broker.log
#停止namesrvsh mqshutdown namesrv
#停止brokersh mqshutdown broker



3.3 常见问题


1.启动失败


启动日志出现启动失败

shutdown hook done, consuming time total(ms): 2114


原因:有可能是内存不够 rocketmq默认内存需要4g(4.7版本)


解决办法:修改启动配置


修改bin目录下的runserver.sh


修改属性

JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx128m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"


修改bin目录下的runbroker.sh

JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx128m -Xmn128m"


3.4简单demo


启动生产者发送消息

#生产者exportNAMESRV_ADDR=localhost:9876
sh tools.sh org.apache.rocketmq.example.quickstart.Producer
#消费者exportNAMESRV_ADDR=localhost:9876
sh tools.sh org.apache.rocketmq.example.quickstart.Consumer


3.5 rocketmq各角色说明


producer: 生产者,消息发送方


consumer: 消费者,消息接收方


Broker: 消息存储和发送服务


NameSrv: 管理Broker地址


Topic: 主题


Message Quene


3.6 集群


3.6.1 集群模式


单Master模式


多Master模式


多Master多Slave模式(异步)


多Master多Slave模式(同步)


3.6.2 双主双从(同步)集群搭建


3.6.2.1 环境配置


简述: 部署到两台虚拟机


192.168.164.128 Master1 Slave2


192.168.164.129 Master2 Slave1


修改host文件 ip和域名映射
vim /etc/hosts
#nameserver192.168.164.128 rocketmq-nameserver1
192.168.164.129 rocketmq-nameserver2
#broker192.168.164.128 rocketmq-master1
192.168.164.128 rocketmq-slave2
192.168.164.129 rockermq-masert2
192.168.164.129 rocketmq-slave1
#重启网卡systemctl restart network



关闭防火墙
systemctl stop firewalld
systemctl status firewalld



配置rocketmq环境变量
vim /etc/profile
#追加exportROCKETMQ_HOME=/home/rocketmq
exportPATH=$PATH:$ROCKETMQ_HOME/bin


建立rocketmq日志目录
#Master1节点mkdir-p /home/rocketmq/store/broker-a
mkdir /home/rocketmq/store/broker-a/commitlog
mkdir /home/rocketmq/store/broker-a/consumequeue
mkdir /home/rocketmq/store/broker-a/index
mkdir /home/rocketmq/store/broker-b-s
mkdir /home/rocketmq/store/broker-b-s/commitlog
mkdir /home/rocketmq/store/broker-b-s/consumequeue
mkdir /home/rocketmq/store/broker-b-s/index
#Master2节点mkdir-p /home/rocketmq/store/broker-b
mkdir /home/rocketmq/store/broker-b/commitlog
mkdir /home/rocketmq/store/broker-b/consumequeue
mkdir /home/rocketmq/store/broker-b/index
mkdir /home/rocketmq/store/broker-a-s
mkdir /home/rocketmq/store/broker-a-s/commitlog
mkdir /home/rocketmq/store/broker-a-s/consumequeue
mkdir /home/rocketmq/store/broker-a-s/index


3.6.2.2 修改双主双从配置文件


Master1 修改配置文件
vim /home/rocketmq/conf/2m-2s-sync/broker-a.properties
# Licensed to the Apache Software Foundation (ASF) under one or more# contributor license agreements.  See the NOTICE file distributed with# this work for additional information regarding copyright ownership.# The ASF licenses this file to You under the Apache License, Version 2.0# (the "License"); you may not use this file except in compliance with# the License.  You may obtain a copy of the License at##     http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an "AS IS" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.# See the License for the specific language governing permissions and# limitations under the License.# brokerClusterName=DefaultCluster# brokerName=broker-a# brokerId=0# deleteWhen=04# fileReservedTime=48# brokerRole=SYNC_MASTER# flushDiskType=ASYNC_FLUSH#所属的集群名字brokerClusterName=rocketmq-cluster
#broker名字 不同的broker填的不一样brokerName=broker-a
#brokerId 0表示Master 非0表示SlavebrokerId=0#nameserver地址 配置多个分号隔开namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#发送消息时默认创建服务器上不存在的Topic,默认创建的队列数defaultTopicQueueNums=4#自动创建TopicautoCreateTopicEnable=true#自动创建订阅组autoCreateSubscriptionGroup=true#对外服务的监听端口listenPort=10911#删除文件时间点 凌晨四点deleteWhen=04#文件保留时间 默认48小时fileReservedTime=120#文件默认大小 1gmapedFileSizeCommitLog=1024000000#每个文件默认存储30w条消息mapedFileSizeConsumeQuene=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/home/rocketmq/store/broker-a
#commitLog存储路径storePathCommitLog=/home/rocketmq/store/broker-a/commitlog
#消息队列存储路径storePathConsumeQueue=/home/rocketmq/store/broker-a/consumequeue
#消息队列索引存储路径storePathIndex=/home/rocketmq/store/broker-a/index
#checkPoint 文件存储路径storeCheckpoint=/home/rocketmq/store/broker-a/checkpoint
#abort文件存储路径abortFile=/home/rocketmq/store/broker-a/abort
#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SYNC_MASTER
#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=SYNC_FLUSH
#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=128



Slave2 修改配置文件
vim /home/rocketmq/conf/2m-2s-sync/broker-b-s.properties
#所属的集群名字brokerClusterName=rocketmq-cluster
#broker名字 不同的broker填的不一样brokerName=broker-b
#brokerId 0表示Master 非0表示SlavebrokerId=1#nameserver地址 配置多个分号隔开namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#发送消息时默认创建服务器上不存在的Topic,默认创建的队列数defaultTopicQueueNums=4#自动创建TopicautoCreateTopicEnable=true#自动创建订阅组autoCreateSubscriptionGroup=true#对外服务的监听端口listenPort=11011#删除文件时间点 凌晨四点deleteWhen=04#文件保留时间 默认48小时fileReservedTime=120#文件默认大小 1gmapedFileSizeCommitLog=1024000000#每个文件默认存储30w条消息mapedFileSizeConsumeQuene=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/home/rocketmq/store/broker-b-s
#commitLog存储路径storePathCommitLog=/home/rocketmq/store/broker-b-s/commitlog
#消息队列存储路径storePathConsumeQueue=/home/rocketmq/store/broker-b-s/consumequeue
#消息队列索引存储路径storePathIndex=/home/rocketmq/store/broker-b-s/index
#checkPoint 文件存储路径storeCheckpoint=/home/rocketmq/store/broker-b-s/checkpoint
#abort文件存储路径abortFile=/home/rocketmq/store/broker-b-s/abort
#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SLAVE
#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=128



Master2 修改配置文件
vim /home/rocketmq/conf/2m-2s-sync/broker-b.properties
# Licensed to the Apache Software Foundation (ASF) under one or more# contributor license agreements.  See the NOTICE file distributed with# this work for additional information regarding copyright ownership.# The ASF licenses this file to You under the Apache License, Version 2.0# (the "License"); you may not use this file except in compliance with# the License.  You may obtain a copy of the License at##     http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an "AS IS" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.# See the License for the specific language governing permissions and# limitations under the License.# brokerClusterName=DefaultCluster# brokerName=broker-a# brokerId=0# deleteWhen=04# fileReservedTime=48# brokerRole=SYNC_MASTER# flushDiskType=ASYNC_FLUSH#所属的集群名字brokerClusterName=rocketmq-cluster
#broker名字 不同的broker填的不一样brokerName=broker-b
#brokerId 0表示Master 非0表示SlavebrokerId=0#nameserver地址 配置多个分号隔开namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#发送消息时默认创建服务器上不存在的Topic,默认创建的队列数defaultTopicQueueNums=4#自动创建TopicautoCreateTopicEnable=true#自动创建订阅组autoCreateSubscriptionGroup=true#对外服务的监听端口listenPort=10911#删除文件时间点 凌晨四点deleteWhen=04#文件保留时间 默认48小时fileReservedTime=120#文件默认大小 1gmapedFileSizeCommitLog=1024000000#每个文件默认存储30w条消息mapedFileSizeConsumeQuene=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/home/rocketmq/store/broker-b
#commitLog存储路径storePathCommitLog=/home/rocketmq/store/broker-b/commitlog
#消息队列存储路径storePathConsumeQueue=/home/rocketmq/store/broker-b/consumequeue
#消息队列索引存储路径storePathIndex=/home/rocketmq/store/broker-b/index
#checkPoint 文件存储路径storeCheckpoint=/home/rocketmq/store/broker-b/checkpoint
#abort文件存储路径abortFile=/home/rocketmq/store/broker-b/abort
#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SYNC_MASTER
#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=SYNC_FLUSH
#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=128



Slave1 修改配置文件
vim /home/rocketmq/conf/2m-2s-sync/broker-a-s.properties
#所属的集群名字brokerClusterName=rocketmq-cluster
#broker名字 不同的broker填的不一样brokerName=broker-a
#brokerId 0表示Master 非0表示SlavebrokerId=1#nameserver地址 配置多个分号隔开namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#发送消息时默认创建服务器上不存在的Topic,默认创建的队列数defaultTopicQueueNums=4#自动创建TopicautoCreateTopicEnable=true#自动创建订阅组autoCreateSubscriptionGroup=true#对外服务的监听端口listenPort=11011#删除文件时间点 凌晨四点deleteWhen=04#文件保留时间 默认48小时fileReservedTime=120#文件默认大小 1gmapedFileSizeCommitLog=1024000000#每个文件默认存储30w条消息mapedFileSizeConsumeQuene=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/home/rocketmq/store/broker-a-s
#commitLog存储路径storePathCommitLog=/home/rocketmq/store/broker-a-s/commitlog
#消息队列存储路径storePathConsumeQueue=/home/rocketmq/store/broker-a-s/consumequeue
#消息队列索引存储路径storePathIndex=/home/rocketmq/store/broker-a-s/index
#checkPoint 文件存储路径storeCheckpoint=/home/rocketmq/store/broker-a-s/checkpoint
#abort文件存储路径abortFile=/home/rocketmq/store/broker-a-s/abort
#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SLAVE
#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=128



3.6.2.3 修改broker启动文件


修改启动配置文件,防止内存要求过大导致无法启动问题

vim /home/rocketmq/bin/runbroker.sh
修改内存
vim /home/rocketmq/bin/runserver.sh
修改内存



3.6.2.4 启动namesrv集群


两台服务器分别执行

nohup sh mqnamesrv &


3.6.2.5 启动broker集群


Master1
nohup sh mqbroker -c /home/rocketmq/conf/2m-2s-sync/broker-a.properties &



Slave2
nohup sh mqbroker -c /home/rocketmq/conf/2m-2s-sync/broker-b-s.properties &



Master2
nohup sh mqbroker -c /home/rocketmq/conf/2m-2s-sync/broker-b.properties &


Slave1
nohup sh mqbroker -c /home/rocketmq/conf/2m-2s-sync/broker-a-s.properties &


3.7 mqadmin


格式: mqadmin command args


3.7.1 Topic


3.8 rocketConsole搭建


下载源码 修改配置application.properties nameserver 地址


mvn编译


java -jar 启动jar包


3.9 消息发送与消费


引入jar包



3.9.1 发送消息基本流程


  1. 创建生产者,指定生产者组名
  2. 指定nameserver地址
  3. 启动producer
  4. 创建消息对象,指定Topic,tag和消息体(内容)
  5. 发送消息
  6. 关闭producer


3.9.2 消费消息基本流程


  1. 创建消费者,指定消费者组名
  2. 指定nameserver地址
  3. 订阅Topic,tag
  4. 设置回调,处理消息
  5. 启动消费者(消费者是一直在监听的)


3.9.3 基本消息


3.9.3.1 消息发送(生产者)


3.9.3.1.1 发送同步消息
//创建生产者对象 指定group 如果group不存在 指定group name可以直接创建,如果存在需////要//setProducerGroupDefaultMQProducerproducer=newDefaultMQProducer();
//指定nameserver 集群分号隔开producer.setNamesrvAddr("192.168.164.128:9876;192.168.164.129;");
producer.setProducerGroup("baseGroup");
//启动producerproducer.start();
//循环发送十条消息for (inti=0; i<10; i++) {
//创建消息对象 指定topic tag 和消息体Messagemessage=newMessage("baseTopic", "baseTag", (("hello world")+(i+1)).getBytes());
//发送消息SendResultsend=producer.send(message);
//打印消息发送后的返回值StringmsgId=send.getMsgId();
//            System.out.println("msgId====="+msgId);SendStatussendStatus=send.getSendStatus();
//            System.out.println("sendStatus====="+sendStatus);intqueueId=send.getMessageQueue().getQueueId();
//            System.out.println("queueId====="+queueId);System.out.println("发送结果"+send);
//线程休眠1sTimeUnit.SECONDS.sleep(1);
}
//关闭producerproducer.shutdown();



3.9.3.1.2 发送异步消息
//创建生产者对象 指定groupDefaultMQProducerproducer=newDefaultMQProducer();
//指定nameserver 集群分号隔开producer.setNamesrvAddr("192.168.164.128:9876;192.168.164.129;");
producer.setProducerGroup("baseGroup");
//启动producerproducer.start();
//循环发送十条消息for (inti=0; i<10; i++) {
//创建消息对象 指定topic tag 和消息体Messagemessage=newMessage("baseTopic", "baseTag", (("hello world")+(i+1)).getBytes());
//发送消息producer.send(message, newSendCallback() {
@OverridepublicvoidonSuccess(SendResultsendResult) {
System.out.println("发送结果"+sendResult);
                }
@OverridepublicvoidonException(Throwablethrowable) {
System.out.println("发送失败");
                }
            });
//线程休眠1sTimeUnit.SECONDS.sleep(1);
        }
//关闭producerproducer.shutdown();

3.9.3.1.3 发送单项消息


无返回结果

//创建生产者对象 指定groupDefaultMQProducerproducer=newDefaultMQProducer();
//指定nameserver 集群分号隔开producer.setNamesrvAddr("192.168.164.128:9876;192.168.164.129;");
producer.setProducerGroup("baseGroup");
//启动producerproducer.start();
//循环发送十条消息for (inti=0; i<10; i++) {
//创建消息对象 指定topic tag 和消息体Messagemessage=newMessage("baseTopic", "baseTag", (("hello world")+(i+1)).getBytes());
producer.sendOneway(message);
//线程休眠1sTimeUnit.SECONDS.sleep(1);
        }
//关闭producerproducer.shutdown();

 

3.9.3.2 消息消费(消费者)


3.9.3.2.1 push方式消费
//创建消费者对象DefaultMQPushConsumerconsumer=newDefaultMQPushConsumer("baseConsumer");
consumer.setNamesrvAddr("192.168.164.128:9876;192.168.164.129:9876");
//订阅topic,tagconsumer.subscribe("baseTopic", "baseTag");
//监听消息consumer.registerMessageListener(newMessageListenerConcurrently() {
@OverridepublicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>list, ConsumeConcurrentlyContextconsumeConcurrentlyContext) {
for (MessageExtmessageExt : list) {
System.out.println(newString(messageExt.getBody()));
                }
returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
consumer.start();
System.out.println("consumer启动");

 


3.9.3.2.2 消费模式


  1. 广播模式 所有订阅Topic的消费者都会消费消息
  2. 负载均衡模式 同一个组的消费者会瓜分Topic消息


3.9.4 顺序消息


业务需要按照顺序发送和消费消息


rocketmq自带顺序消息


3.9.4.1 原理


使用队列的数据结构实现顺序消息   先进先出


生产消息时把消息发送到broker的同一队列


消费消息时同一个队列的消息由同一个线程处理


3.9.4.2 生产顺序消息

try {
DefaultMQProducerproducer=newDefaultMQProducer();
producer.setProducerGroup("orderGroup");
producer.setNamesrvAddr("192.168.164.128:9876;192.168.164.129:9876");
List<OrderEntity>list=OrderEntity.builds();
for (OrderEntityo : list) {
Messagemessage=newMessage("orderTopic", "orderTag", o.toString().getBytes());
SendResultsend=producer.send(message, newMessageQueueSelector() {
//指定消息发送使用的队列,同一个orderId的消息统一发送到一个queue中@OverridepublicMessageQueueselect(List<MessageQueue>list, Messagemessage, Objecto) {
longl= (Long) o%list.size();
returnlist.get((int)l);
            }
        }, o.getOrderId());
System.out.println(send);
    }
}catch (Exceptione){
e.printStackTrace();
}finally {
producer.shutdown();
}



3.9.4.3 消费顺序消息

DefaultMQPushConsumerconsumer=newDefaultMQPushConsumer();
consumer.setConsumerGroup("orderGroup");
consumer.setInstanceName("orderConsumer");
try {
consumer.subscribe("orderTopic", "*");
} catch (MQClientExceptione) {
e.printStackTrace();
}
consumer.setNamesrvAddr("192.168.164.128:9876;192.168.164.129:9876");
//注册一个顺序消息的监听器defaultMQPushConsumer.registerMessageListener(newMessageListenerOrderly(){
@OverridepublicConsumeOrderlyStatusconsumeMessage(List<MessageExt>list, ConsumeOrderlyContextconsumeOrderlyContext) {
for (inti=0; i<list.size(); i++) {
System.out.println(newString(list.get(i).getBody()));
        }
returnConsumeOrderlyStatus.SUCCESS;
    }
});
try {
defaultMQPushConsumer.start();
} catch (MQClientExceptione) {
e.printStackTrace();
}



3.9.5 延时消息


messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h


1就是1s


2就是5s


...


共18个级别


3.9.5.1 发送消息



//核心代码message.setDelayTimeLevel(2);
for (inti=0; i<10; i++) {
//创建消息对象 指定topic tag 和消息体Messagemessage=newMessage("orderTopic", "orderTag", (("hello world")+(i+1)).getBytes());
//设置延时消息message.setDelayTimeLevel(3);
//发送消息System.out.println("开始时间==="+System.currentTimeMillis());
SendResultsend=producer.send(message);
//打印消息发送后的返回值StringmsgId=send.getMsgId();
//            System.out.println("msgId====="+msgId);SendStatussendStatus=send.getSendStatus();
//            System.out.println("sendStatus====="+sendStatus);intqueueId=send.getMessageQueue().getQueueId();
//            System.out.println("queueId====="+queueId);System.out.println("发送结果"+send);
//线程休眠1sTimeUnit.SECONDS.sleep(1);
 }


3.9.5.2 消费消息


正常消费


3.9.6 批量消息


3.9.6.1 生产消息

Messagemessage=newMessage("orderTopic", "orderTag", (("hello world")+ (2)).getBytes());
Messagemessage1=newMessage("orderTopic", "orderTag", (("hello world")+(3)).getBytes()); 
Messagemessage2=newMessage("orderTopic", "orderTag", (("hello world")+(4)).getBytes());
Messagemessage3=newMessage("orderTopic", "orderTag", (("hello world")+(5)).getBytes());
List<Message>msgs=newArrayList<>();
msgs.add(message);
msgs.add(message1);
msgs.add(message2);
msgs.add(message3);
///发送消息System.out.println("开始时间==="+System.currentTimeMillis());
SendResultsend=producer.send(msgs);


3.9.6.2 消费消息


正常消费


3.9.7 过滤消息


3.9.7.1 tag过滤


在订阅tag时可以通过*或者||指定订阅的tag

consumer.subscribe("baseTopic", "baseTag || orderTag");
consumer.subscribe("baseTopic", "*");


3.9.7.2 sql过滤


生产消息时可以给消息指定一个特定的属性

Messagemessage=newMessage("orderTopic", "orderTag", (("hello world") + (2)).getBytes());
message.putUserProperty("type", i+"");



这样在消费者订阅消息时可以通过简单sql过滤

consumer.subscribe("orderTopic", MessageSelector.bySql("type>1"));


批量消息不能使用过滤


3.9.8 事务消息



3.9.8.1 概述


具体查看


发送事务消息时有三种状态


  1. 提交 提交后消息才能被消费
  2. 回滚 回滚后消息被删除
  3. 中间 没有提交或回滚的状态,这个状态下rocketmq会去回查本地事务,确定要提交还是回滚


3.9.8.2 发送消息

TransactionMQProducertransactionMQProducer=newTransactionMQProducer("transactionGroup");
transactionMQProducer.setNamesrvAddr("192.168.164.128:9876;192.168.164.129;");
transactionMQProducer.setInstanceName("transactionProducer");
transactionMQProducer.setTransactionListener(newTransactionListener() {
//发送半消息后触发,是提交还是回滚@OverridepublicLocalTransactionStateexecuteLocalTransaction(Messagemessage, Objecto) {
System.out.println("message: "+message.getUserProperty("type")+"开始检查事务状态");
if(StringUtils.equals("commit",message.getUserProperty("type"))){
returnLocalTransactionState.COMMIT_MESSAGE;
                }elseif (StringUtils.equals("rollback",message.getUserProperty("type"))){
returnLocalTransactionState.ROLLBACK_MESSAGE;
                }else{
returnLocalTransactionState.UNKNOW;
                }
            }
//当消息状态处于中间状态时,rocketmq会回调此方法,验证此消息是发送还是回滚@OverridepublicLocalTransactionStatecheckLocalTransaction(MessageExtmessageExt) {
System.out.println("message: "+messageExt.getMsgId()+"触发回查");
returnLocalTransactionState.COMMIT_MESSAGE;
            }
        });
transactionMQProducer.start();
Messagemessage=newMessage("transactionTopic", "transcationTag", "hello world message0".getBytes());
message.putUserProperty("type", "commit");
Messagemessage1=newMessage("transactionTopic", "transcationTag", "hello world message1".getBytes());
message1.putUserProperty("type", "rollback");
Messagemessage2=newMessage("transactionTopic", "transcationTag", "hello world message2".getBytes());
SendResultsend=transactionMQProducer.sendMessageInTransaction(message,null);
System.out.println("发送结果=="+send);
Thread.sleep(2000);
SendResultsend1=transactionMQProducer.sendMessageInTransaction(message1,null);
System.out.println("发送结果=="+send1);
Thread.sleep(2000);
SendResultsend2=transactionMQProducer.sendMessageInTransaction(message2,null);
System.out.println("发送结果=="+send2);
Thread.sleep(100000);
transactionMQProducer.shutdown();



3.9.8.3 限制


1.不支持延时消息和批量消息


2.事务消息可能被多次回查,可能出现重复消费情况


3.通过设置用户属性来设置服务端的回查间隔时间

message2.putUserProperty(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS,"5");


文章标签：
云消息队列 MQ
日志服务
Java
消息中间件
SQL
负载均衡
网络安全
RocketMQ
Kafka
Scala
存储
大数据
关键词：
云消息队列 MQ消息中间件
消息中间件云消息队列 MQ
中间件rocketmq
消息队列RocketMQ版消息中间件
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
z.y.s
目录
相关文章
菜菜的后端私房菜
|
2月前
|
消息中间件 存储 Java
RocketMQ（一）：消息中间件缘起，一览整体架构及核心组件
【10月更文挑战第15天】本文介绍了消息中间件的基本概念和特点，重点解析了RocketMQ的整体架构和核心组件。消息中间件如RocketMQ、RabbitMQ、Kafka等，具备异步通信、持久化、削峰填谷、系统解耦等特点，适用于分布式系统。RocketMQ的架构包括NameServer、Broker、Producer、Consumer等组件，通过这些组件实现消息的生产、存储和消费。文章还提供了Spring Boot快速上手RocketMQ的示例代码，帮助读者快速入门。
菜菜的后端私房菜
163 6
1176112968452250
|
3月前
|
消息中间件 存储 RocketMQ
消息中间件-RocketMQ技术(二）
消息中间件-RocketMQ技术(二）
1176112968452250
64 1
1176112968452250
|
3月前
|
消息中间件 存储 中间件
消息中间件-RocketMQ技术(一）
消息中间件-RocketMQ技术(一）
1176112968452250
66 1
m1a6
|
27天前
|
消息中间件 存储 Apache
探索 RocketMQ：企业级消息中间件的选择与应用
RocketMQ 是一个高性能、高可靠、可扩展的分布式消息中间件，它是由阿里巴巴开发并贡献给 Apache 软件基金会的一个开源项目。RocketMQ 主要用于处理大规模、高吞吐量、低延迟的消息传递，它是一个轻量级的、功能强大的消息队列系统，广泛应用于金融、电商、日志系统、数据分析等领域。
m1a6
55 0
探索 RocketMQ：企业级消息中间件的选择与应用
江湖有缘
|
2月前
|
消息中间件 编解码 Docker
【Docker项目实战】Docker部署RabbitMQ消息中间件
【10月更文挑战第8天】Docker部署RabbitMQ消息中间件
江湖有缘
122 1
【Docker项目实战】Docker部署RabbitMQ消息中间件
mikechen的互联网架构
|
1月前
|
消息中间件 存储 Java
吃透 RocketMQ 消息中间件，看这篇就够了！
本文详细介绍 RocketMQ 的五大要点、核心特性及应用场景，涵盖高并发业务场景下的消息中间件关键知识点。关注【mikechen的互联网架构】，10年+BAT架构经验倾囊相授。
mikechen的互联网架构
145 0
吃透 RocketMQ 消息中间件，看这篇就够了！
wei_shuo
|
7月前
|
消息中间件 存储 负载均衡
消息中间件的选择：RabbitMQ是一个明智的选择
消息中间件的选择：RabbitMQ是一个明智的选择
wei_shuo
120 0
_BugMan
|
6月前
|
消息中间件 存储 中间件
【消息中间件】详解三大MQ：RabbitMQ、RocketMQ、Kafka
【消息中间件】详解三大MQ：RabbitMQ、RocketMQ、Kafka
_BugMan
1644 0
江湖有缘
|
5月前
|
消息中间件 编解码 Docker
Docker部署RabbitMQ消息中间件
【7月更文挑战第4天】Docker部署RabbitMQ消息中间件
江湖有缘
280 3
热爱技术的小郑
|
4月前
|
消息中间件 Java 测试技术
消息中间件RabbitMQ---SpringBoot整合RabbitMQ【三】
这篇文章是关于如何在SpringBoot应用中整合RabbitMQ的消息中间件。内容包括了在SpringBoot项目中添加RabbitMQ的依赖、配置文件设置、启动类注解，以及如何通过单元测试来创建交换器、队列、绑定，并发送和接收消息。文章还介绍了如何配置消息转换器以支持对象的序列化和反序列化，以及如何使用注解`@RabbitListener`来接收消息。
热爱技术的小郑
71 2
消息中间件RabbitMQ---SpringBoot整合RabbitMQ【三】
云原生

中间件

热门文章

最新文章

  • 1
    3分钟，了解阿里云热门开发者工具 Cloud Toolkit
  • 2
    Java编码技巧之高效代码50例
  • 3
    开发者说：Sentinel 流控功能在 SpringMVC/SpringBoot 上的实践
  • 4
    Kafka、RabbitMQ、RocketMQ 消息中间件的对比 | 消息发送性能篇
  • 5
    分布式服务架构下的混沌工程实践
  • 6
    新的开始 | Arthas GitHub Star 破万后的回顾和展望
  • 7
    应用高可用 AHAS 一键提升云上的业务可用性
  • 8
    解读OpenMessaging开源项目，阿里巴巴发起首个分布式消息领域的国际标准
  • 9
    网站常见问题1分钟定位 - 如何使用阿里云ARMS轻松重现用户浏览器问题
  • 10
    云栖开发者沙龙 - 数据库分库分表中间件(杭州站) 业内专家大咖齐聚，期待你的参与
  • 1
    杭州铭师堂的云原生升级实践
    31
  • 2
    活动速递丨AI 原生应用开发实战营·上海站邀您参会
    17
  • 3
    MSE Nacos 2.3.2.0 发布，性能最多提升三倍，支持操作审计等安全特性
    29
  • 4
    邀您参加云原生高可用技术沙龙丨云上高可用体系构建：从理论到实践
    29
  • 5
    Nacos 3.0 Alpha 发布，在安全、泛用、云原生更进一步
    28
  • 6
    云上数据安全保护：敏感日志扫描与脱敏实践详解
    36
  • 7
    SpringCloud 应用 Nacos 配置中心注解
    65
  • 8
    （前端3D模型开发）网页三维CAD中加载和保存STEP模型
    52
  • 9
    微服务引擎 MSE 及云原生 API 网关 2024 年 11 月产品动态
    46
  • 10
    两招玩转阿里云系统事件监控
    47

    • 相关产品

  • 云消息队列 MQ
    文档详情 产品详情

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • 金融分布式中间件分享
  • 无需部署的前端中间件技术——企业移动化新思路
  • 《云原生中间件精品案例集》

    • 相关实验场景

    更多
  • 快速体验云消息队列RocketMQ版
  • 快速入门轻量消息队列（原MNS）
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 消息队列RocketMQ版：基础消息收发功能体验
    • 下一篇
    DataWorks