Q概念用途特性安装RocketMQ掌握RocketMQ的api使用对producer、consumer进行详解了解RocketMQ的存储特点

简介及相关概念JavaAPISpringBoot整合RocketMQ消息的顺序收发消息系统的事务、存储、重试策略消息系统的集群

RocketMQ

阿里中间件Notify用于交易核心信息的流转2010年B2B开始大规模使用ActiveMQ作为消息内核急需支持顺序消息、拥有海量消息堆积能力的消息中间件——MetaQ

1.0

年MetaQ发展到了3.0版本抽象除了通用的消息引擎RorcketMQ2015年RocketMQ进过双十一在可用性可靠性和稳定性等方面都有出色表现。

阿里消息中间件基于RocketMQ退出Aliware

MQ1.0开始为阿里云上的企业提供消息服务2016年RocketMQ进入Apache孵化

Producer

Group一类Producer的集合名称这类Producer通常发送同一类消息且发送逻辑一致

Consumer

Group一类Consumer的集合名称这类Producer通常发送同一类消息且发送逻辑一致

Broker

RocketMQ的核心消息的发送、接收、高可用等需要定时发送自身情况到NameServer,默认10s发送一次超过2分钟会认为该broker失效

NameServer

不同类型的消息以不同的Topic名称进行区分如User、Order等Message

Queue

下载地址https://archive.apache.org/dist/rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip

/opt

rocketmq-all-4.3.2-bin-release.zip

rocketmq-all-4.3.2-bin-release/#

启动nameserver

os::commit_memory(0x00000005c0000000,

8589934592,

(errno12)启动错误因为RocketMQ的配置默认是生产环境的配置设置jvm的内存值比较大需要调整默认值

bin/

broker[iZ2zeg4pktzjhp9h7wt6doZ,

172.17.0.1:10911]

/opt/rocketmq-all-4.3.2-bin-release/bin

tools.sh

org.apache.rocketmq.example.quickstart.Producer接收消息测试

tools.sh

org.apache.rocketmq.example.quickstart.Consumerjava

api测试

dependenciesdependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client/artifactIdversion4.3.2/version/dependency

/dependenciesbuildpluginsplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-compiler-plugin/artifactIdversion3.2/versionconfigurationsource1.8/sourcetarget1.8/targetencodingUTF-8/encoding/configuration/plugin/plugins

/build测试代码

org.apache.rocketmq.client.exception.MQBrokerException;

import

org.apache.rocketmq.client.exception.MQClientException;

import

org.apache.rocketmq.client.producer.DefaultMQProducer;

import

org.apache.rocketmq.client.producer.SendResult;

import

org.apache.rocketmq.common.message.Message;

import

org.apache.rocketmq.remoting.common.RemotingHelper;

import

org.apache.rocketmq.remoting.exception.RemotingException;import

java.io.UnsupportedEncodingException;public

class

DefaultMQProducer(test-group);//specify

name

addressproducer.setNamesrvAddr(8.140.130.91:9876);//Lanuch

the

Message(TopicTest1,/*topic*/TAGA,/*tag*/(Hello

RocketMQ

i).getBytes(RemotingHelper.DEFAULT_CHARSET)/*message

body*/);//Call

producer.send(msg);System.out.printf(%s%n,

sendResult);}//Shut

broker的ip地址是172.17.0.1,为私有ip所以不可访问

/opt/rocketmq-all-4.3.2-bin-release/conf

vim

broker.confbrokerIP18.140.130.91

brokerNamebroker_haoke_im#启动broker通过

指定配置文件

/opt/rocketmq-all-4.3.2-bin-release/

bin/mqbroker

/opt/rocketmq-all-4.3.2-bin-release/conf/broker.confAPI测试成功

#拉取镜像

foxiswho/rocketmq:broker-4.3.2#创建nameserver容器

docker

/data/rmq-data/rmqserver/logs:/opt/logs

/data/rmq-data/rmqserver/store:/opt/store

foxiswho/rocketmq:server-4.3.2#创建broker容器

#10911

/data/rmq-data/rmqbroker/conf/broker.conf:/etc/rocketmq/broker.conf

/data/rmq-data/rmqbroker/logs:/opt/logs

/data/rmq-data/rmqbroker/store:/opt/store

foxiswho/rocketmq:broker-4.3.2#启动容器

docker

enablePropertyFiltertrue部署RocketMQ的管理工具

UI管理工具rocketmq-console,项目地址https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console

#拉取镜像

apacherocketmq/rocketmq-console:2.0.0#创建并启动容器

docker

JAVA_OPTS-Drocketmq.config.namesrvAddr8.140.130.91:9876

-Drocketmq.config.isVIPChannelfalse

8082:8080

apacherocketmq/rocketmq-console:2.0.0访问http://8.140.130.91:8082/

Java

org.apache.rocketmq.client.producer.DefaultMQProducer;public

class

Exception{//设置NameServer地址DefaultMQProducer

producer

DefaultMQProducer(test-group);//设置producer

的NameServerAddressproducer.setNamesrvAddr(8.140.130.91:9876);//启动NameServerproducer.start();/**

创建topic*

*/producer.createTopic(broker_haoke_im,test_topic,8);System.out.println(topic创建成功);producer.shutdown();}

}发送消息

值说明Topicnull必填线下环境不需要申请线上环境需要申请后才能使用Bodynull必填二进制形式序列化由应用决定Producer

Consumer

会延时特定的时间才会被消费WaitStoreMsgOKTRUE选填表示消息是否在服务器落盘后才返回应答。

package

org.apache.rocketmq.client.producer.DefaultMQProducer;

import

org.apache.rocketmq.client.producer.SendResult;

import

org.apache.rocketmq.common.message.Message;public

class

DefaultMQProducer(test-group);producer.setNamesrvAddr(8.140.130.91:9876);producer.start();String

msgStr

Message(test_topic,test,msgStr.getBytes(UTF-8));SendResult

result

producer.send(message);System.out.println(result);System.out.println(消息状态

result.getSendStatus());System.out.println(消息id

result.getMsgId());System.out.println(消息queue

result.getMessageQueue());System.out.println(消息offset

result.getQueueOffset());producer.shutdown();}

}异步

org.apache.rocketmq.client.producer.DefaultMQProducer;

import

org.apache.rocketmq.client.producer.SendCallback;

import

org.apache.rocketmq.client.producer.SendResult;

import

org.apache.rocketmq.common.message.Message;public

class

DefaultMQProducer(test-group);producer.setNamesrvAddr(8.140.130.91:9876);producer.start();String

msgStr

Message(test_topic,test,msgStr.getBytes(UTF-8));producer.send(message,

new

{System.out.println(result);System.out.println(消息状态

result.getSendStatus());System.out.println(消息id

result.getMsgId());System.out.println(消息queue

result.getMessageQueue());System.out.println(消息offset

result.getQueueOffset());}Overridepublic

void

{System.out.println(消息发送失败);}});//

producer.shutdown()要注释掉否则发送失败。

原因是异步发送还未来得及发送就被关闭了//producer.shutdown();}

}消费信息

org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

import

org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;

import

org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;

import

org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;

import

org.apache.rocketmq.common.message.MessageExt;import

java.io.UnsupportedEncodingException;

import

DefaultMQPushConsumer(test-group);consumer.setNamesrvAddr(8.140.130.91:9876);//订阅topic接收此topic下的所有消息consumer.subscribe(test_topic,*);consumer.registerMessageListener(new

msgs,

{e.printStackTrace();}}System.out.println(收到消息-msgs);/**

CONSUME_SUCCESS

ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();}

}测试接收历史消息

Message(test_topic,add,msgStr.getBytes(UTF-8));#消费者

//完整匹配

consumer.subscribe(test_topic,add);

//或匹配

consumer.subscribe(test_topic,add

类似与SQL

org.apache.rocketmq.client.producer.DefaultMQProducer;

import

org.apache.rocketmq.client.producer.SendResult;

import

org.apache.rocketmq.common.message.Message;/***

author

example-roketmq-com.rocketmq.filter*/

public

DefaultMQProducer(test-group);producer.setNamesrvAddr(8.140.130.91:9876);producer.start();String

msgStr

Message(test_topic,test,msgStr.getBytes(UTF-8));msg.putUserProperty(age,18);msg.putUserProperty(sex,女);SendResult

result

producer.send(msg);System.out.println(消息状态result.getSendStatus());System.out.println(消息id

result.getMsgId());System.out.println(消息queueresult.getMessageQueue());System.out.println(消息offsetresult.getQueueOffset());producer.shutdown();}

}消息接收方

org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

import

org.apache.rocketmq.client.consumer.MessageSelector;

import

org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;

import

org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;

import

org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;

import

org.apache.rocketmq.client.exception.MQClientException;

import

org.apache.rocketmq.common.message.MessageExt;import

java.io.UnsupportedEncodingException;

import

DefaultMQPushConsumer(test-group);consumer.setNamesrvAddr(8.140.130.91:9876);consumer.subscribe(test_topic,

AND

sex女));consumer.registerMessageListener(new

msgs,

{e.printStackTrace();}}System.out.println(收到消息-msgs);return

ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();}

}测试

Message(test_topic,test,msgStr.getBytes(UTF-8));

msg.putUserProperty(sex,女);可以接收到消息

生产者发送的顺序消息都要放在同一消息队列中才能保证被顺序取出消费者接收的顺序消息需要从同一队列中获取

生产者

org.apache.rocketmq.client.producer.DefaultMQProducer;

import

org.apache.rocketmq.client.producer.SendResult;

import

org.apache.rocketmq.common.message.Message;public

class

DefaultMQProducer(test-group);producer.setNamesrvAddr(8.140.130.91:9876);producer.start();for

(int

10;//生产10个订单的消息,每个订单10条消息String

msgStr

Message(test_topic,ORDER_MSG,msgStr.getBytes(UTF-8));/**

public

producer.send(message,(mqs,msg,arg)-{//匿名函数的作用为选择消息队列的idInteger

(Integer)

mqs.get(index);},//arg与orderId对应orderId);System.out.println(sendResult);}producer.shutdown();}

}消费者

Exception{DefaultMQPushConsumer

consumer

DefaultMQPushConsumer(test-order-group);consumer.setNamesrvAddr(8.140.130.91:9876);consumer.subscribe(test_order_topic,*);consumer.registerMessageListener(new

ConsumeOrderlyStatus

{System.out.println(Thread.currentThread().getName()

msg.getQueueId()

ConsumeOrderlyStatus.SUCCESS;}});consumer.start();}

}可见订单id为3的消息会存入同一消息队列故在同一消息队列的消息可被同一消费线程监听

分布式事务分类

基于单个JVM数据库分库分表基于多个JVM服务拆分基于多JVM服务拆分且数据库分库分表

Half(Prepare)

消息系统暂时不能投递的消息发送方将消息发送到了MQ服务端。

MQ服务端未收到生产者对消息的二次确认此时该消息被标记为

暂不能投递状态

由于网络闪断、生产者应用重启等原因导致某条事务消息的二次确认丢失MQ服务端发现某条消息长期处于

发送方向MQ服务端发送消息

Server将消息持久化成功后向发送方ACK确认消息已经发送成功此时消息为

半消息

Exception{TransactionMQProducer

producer

TransactionMQProducer(test_transaction_producer);producer.setNamesrvAddr(8.140.130.91:9876);//设置事务监听器producer.setTransactionListener(new

TransactionImpl());producer.start();//发送消息Message

message

Message(pay_topic,用户A给用户B转钱.getBytes(UTF-8));producer.sendMessageInTransaction(message,null);Thread.sleep(99999);producer.shutdown();}

}本地事务处理

executeLocalTransaction(Message

msg,

{Thread.sleep(500);System.out.println(用户A账户减500);//

System.out.println(1/0);System.out.println(用户B账户加500元.);Thread.sleep(800);//二次提交确认STATE_MAP.put(msg.getTransactionId(),LocalTransactionState.COMMIT_MESSAGE);return

LocalTransactionState.COMMIT_MESSAGE;}

catch

{e.printStackTrace();}//回滚STATE_MAP.put(msg.getTransactionId(),

LocalTransactionState.ROLLBACK_MESSAGE);return

LocalTransactionState.ROLLBACK_MESSAGE;}/***

消息回查*

checkLocalTransaction(MessageExt

msg)

STATE_MAP.get(msg.getTransactionId());}

}消费者

DefaultMQPushConsumer(test_transaction_consumer);consumer.setNamesrvAddr(8.140.130.91:9876);//订阅topic接收消息consumer.subscribe(pay_topic,*);consumer.registerMessageListener(new

{Overridepublic

ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();}

}测试

executeLocalTransaction(Message

msg,

{System.out.println(用户A账户减500);Thread.sleep(500);//

System.out.println(1/0);System.out.println(用户B账户加500元.);Thread.sleep(800);//二次提交确认STATE_MAP.put(msg.getTransactionId(),LocalTransactionState.COMMIT_MESSAGE);return

return

LocalTransactionState.COMMIT_MESSAGE;}

catch

{e.printStackTrace();}//回滚STATE_MAP.put(msg.getTransactionId(),

LocalTransactionState.ROLLBACK_MESSAGE);return

LocalTransactionState.ROLLBACK_MESSAGE;}/***

消息回查*

checkLocalTransaction(MessageExt

msg)

STATE_MAP.get(msg.getTransactionId()));return

STATE_MAP.get(msg.getTransactionId());}

}Consumer

push模式客户端与服务端建立连接后当服务端有消息将消息推送到客户端pull模式客户端不断的轮询请求服务端来获取新的而消息

push模式需要消息系统与消费端之间建立长连接对消息系统是很大的负担所以在具体实现时都采用消费端主动拉取的方式即consumer轮询从broker拉取消息

将轮询过程都封装了并注册MessageListener监听器取到消息后唤醒MessageListener监听器的consumeMessage()来消费对用户而言感觉消息是被推送来的。

Pull取消息过程需要自己写首先从目标topic中拿到MessageQueue集合并遍历然后针对每个MessageQueue批量取消息。

一次Pull都要记录该队列的offset,知道去完MessageQueue再换另一个

长轮询长连接轮询客户端像传统轮询一样从服务端请求数据服务端会阻塞请求不会立刻返回直到有数据或超时才返回给客户端然后关闭连接客户端处理完响应信息后再向服务器发送新的请求

消息模式

DefaultMQPushConsumer实现了自动保存offset值及多个consumer的负载均衡

//设置组名

DefaultMQPushConsumer(HAOKE_IM);通过

groupname

将多个consumer组合在一起会存在消息的分配问题消息是发送到组还是每个消费者

集群模式默认

同一个ConsumerGroup里的每个Consumer只消费所订阅消息的一部分内容同一个ConsumerGroup里所有消费的内容合起来才是所订阅Topic内容的整体从而达到负载均衡的目的

广播模式

同一个ConsumerGroup里的每个Consumer都能消费到所订阅Topic的全部消息一个消息会被分发多次被多个Consumer消费

集群模式

consumer.setMessageModel(MessageModel.CLUSTERING);

广播模式

consumer.setMessageModel(MessageModel.BROADCASTING);重复消息的解决方案

生产者不断发送重复消息到消息系统

保证每条消息都有唯一编号且保证消息处理成功与去重的日志同时出现

利用一张日志表来记录已经处理成功的消息的ID如果新到的消息ID已经在日志表中那么就不再处理这条消息

如果由消息系统来实现的话肯定会对消息系统的吞吐量和高可用有影响所以最好还是由业务端自己处理消息重复的问题这也是

RocketMQ存储

在RocketMQ中消息数据是保存在磁盘文件中的使用RocketMQ尽可能保证顺序写入比随机写入效率高很多

Consume

是一个逻辑队列存储了这个Queue在CommitLog中的其实offset、log大小和MessageTag的hashcode

每次读取消息队列先读取ConsumerQueue然后再通过consumerQueue中拿到消息主体

同步刷盘和异步刷盘

RocketMQ为提高性能会尽可能保证磁盘的顺序读写。

消息通过Producer写入RocketMQ的时候有两种写磁盘方式分别是同步刷盘与异步刷盘

同步刷盘——安全性

在返回写成功状态时消息已经写入磁盘执行流程消息写入内存的PAGECACHE后立刻通知刷盘线程刷盘等待刷盘完成刷盘线程执行完成后唤醒等待的线程返回消息写成功的状态

异步刷盘——效率

在返回写成功状态时消息可能只是被写入内存的PAGECACHE写操作的返回快吞吐量大当内存里的消息积累到一定程度统一触发写磁盘动作快速写入

修改刷盘方式

在消息的发送和消费过程中都有可能出现错误如网络异常等出现了错误就需要进行错误重试这种消息的重试分为

consumer端重试

producer.setRetryTimesWhenSendFailed(3);//

发送消息,并且指定超时时间

1000);只有同步生产者才会进行错误重试。

只有特定异常才会重试设置的超时时间小于实际执行时间则不会进行重试

//设置发送总次数

this.defaultMQProducer.getRetryTimesWhenSendFailed()

1;for

ResponseCode.TOPIC_NOT_EXIST:case

ResponseCode.SERVICE_NOT_AVAILABLE:case

ResponseCode.NO_PERMISSION:case

ResponseCode.NOT_IN_CURRENT_UNIT:continue;}}

}consumer端重试

消息正常到了消费者端处理失败发生异常。

eg反序列化失败消息数据本身无法处理

消息状态

org.apache.rocketmq.client.consumer.listener;public

enum

consumption*/CONSUME_SUCCESS,/***

Failure

DefaultMQPushConsumer(test_consumer_group);consumer.setNamesrvAddr(8.140.130.91:9876);//

订阅topic接收此Topic下的所有消息consumer.subscribe(test_error_topic,

*);consumer.registerMessageListener(new

{Overridepublic

msgs,ConsumeConcurrentlyContext

context)

{e.printStackTrace();}}System.out.println(收到消息-

msgs);if(msgs.get(0).getReconsumeTimes()

3){//

ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}return

ConsumeConcurrentlyStatus.RECONSUME_LATER;}});consumer.start();}

timeout

风险较大一旦Broker重启或者宕机会导致整个服务不可用只做开发环境

多Master

一个集群无Slave全是Master例如2个Master或者3个Master单台机器宕机这台机器上未被消费的消息在机器恢复之前不可订阅消息的实时性受到影响

多Master多Slave异步复制

每个Master配置一个Slave有多个Master-Slave对HA双机集群系统采用异步复制方式主备有短暂消息延迟毫秒级优点即使磁盘损坏丢失的消息非常少实时性不会收到影响消费者仍可从Slave消费此过程对应用透明不需人工干预性能同多Master模式一样缺点Master宕机或磁盘损坏会丢失少量消息

多Master多Slave,同步双写

每个Master配置一个Slave有多个Master-Slave对HA双机集群系统采用同步双写方式主备都写成功向应用返回成功优点数据与服务无单点Master宕机情况下消息无延迟服务可用性和数据可用性非常高缺点性能比异步复制模式低

搭建2m2s集群

/data/rmq-data/rmqserver01/logs:/opt/logs

/data/rmq-data/rmqserver01/store:/opt/store

foxiswho/rocketmq:server-4.3.2#nameserver2

docker

/data/rmq-data/rmqserver02/logs:/opt/logs

/data/rmq-data/rmqserver02/store:/opt/store

foxiswho/rocketmq:server-4.3.2搭建broker(2master)

#broker01配置文件

namesrvAddr8.140.130.91:9876;8.140.130.91:9877

brokerNamebroker01

/data/rmq-data/rmqbroker01/conf/broker.conf:/etc/rocketmq/broker.conf

/data/rmq-data/rmqbroker01/logs:/opt/logs

/data/rmq-data/rmqbroker01/store:/opt/store

foxiswho/rocketmq:broker-4.3.2brokerId0表示主0表示Slave

单位——h

namesrvAddr8.140.130.91:9876;8.140.130.91:9877

brokerNamebroker02

/data/rmq-data/rmqbroker02/conf/broker.conf:/etc/rocketmq/broker.conf

/data/rmq-data/rmqbroker02/logs:/opt/logs

/data/rmq-data/rmqbroker02/store:/opt/store

foxiswho/rocketmq:broker-4.3.2搭建从broker(slave)

#slave

namesrvAddr8.140.130.91:9876;8.140.130.91:9877

brokerNamebroker01

/data/rmq-data/rmqbroker03/conf/broker.conf:/etc/rocketmq/broker.conf

/data/rmq-data/rmqbroker03/logs:/opt/logs

/data/rmq-data/rmqbroker03/store:/opt/store

foxiswho/rocketmq:broker-4.3.2#slave

broker02配置文件

namesrvAddr8.140.130.91:9876;8.140.130.91:9877

brokerNamebroker02

/data/rmq-data/rmqbroker04/conf/broker.conf:/etc/rocketmq/broker.conf

/data/rmq-data/rmqbroker04/logs:/opt/logs

/data/rmq-data/rmqbroker04/store:/opt/store

foxiswho/rocketmq:broker-4.3.2#启动容器

docker

DefaultMQProducer(test_cluster_group);producer.setNamesrvAddr(8.140.130.91:9876;8.140.130.91:9877);producer.start();String

msgStr

Message(test_cluster_topic,CLUSTER,msgStr.getBytes(UTF-8));SendResult

result

producer.send(message);System.out.println(result);System.out.println(消息状态

result.getSendStatus());System.out.println(消息id

result.getMsgId());System.out.println(消息queue

result.getMessageQueue());System.out.println(消息offset

result.getQueueOffset());producer.shutdown();}

}消费者

DefaultMQPushConsumer(test_cluster_group);consumer.setNamesrvAddr(8.140.130.91:9876;8.140.130.91:9877);//订阅topopic接收此topic下的所有消息consumer.subscribe(test_cluster_topic,*);consumer.registerMessageListener(new

msgs,

{e.printStackTrace();}}System.out.println(收到消息-msgs);/**

CONSUME_SUCCESS

ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();}

下载依赖

由于rocketMQ没有发布到Mven中央仓库需要自行下载源码并载入到本地Maven仓库

#源码地址

https://hub.fastgit.org/apache/rocketmq-spring#进入源码目录执行

mvn

parentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.4.3/version

/parentdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-spring-boot-starter/artifactIdversion2.0.0/version/dependencydependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client/artifactIdversion4.3.2/version/dependencydependencygroupIdjunit/groupIdartifactIdjunit/artifactIdscopetest/scope/dependency

/dependenciesbuildpluginsplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-compiler-plugin/artifactIdversion3.2/versionconfigurationsource1.8/sourcetarget1.8/targetencodingUTF-8/encoding/configuration/plugin/plugins

#Spring

spring.rocketmq.nameServer8.140.130.91:9876

spring.rocketmq.producer.grouptest_spring_producer_group基本使用

生产者发送消息

{//注入rocketmq模板Autowiredprivate

RocketMQTemplate

msg){this.rocketMQTemplate.convertAndSend(topic,msg);}

}启动类

com.rocketmq;SpringBootApplication

public

{SpringApplication.run(MyApplication.class,args);}

}测试生产消息

org.springframework.beans.factory.annotation.Autowired;

import

org.springframework.boot.test.context.SpringBootTest;

import

org.springframework.test.context.junit4.SpringRunner;RunWith(SpringRunner.class)

SpringBootTest

消息;this.producer.sendMsg(test_spring_topic,msg);System.out.println(发送成功!);}}消费者消费消息

package

org.apache.rocketmq.spring.annotation.ConsumeMode;

import

org.apache.rocketmq.spring.annotation.RocketMQMessageListener;

import

org.apache.rocketmq.spring.core.RocketMQListener;

import

org.springframework.stereotype.Component;Component

test_spring_topic,consumerGroup

test_spring_consumer_group,selectorExpression

*,consumeMode

{System.out.println(收到消息-msg);}

}事务消息

com.rocketmq.spring.transaction;Component

public

MessageBuilder.withPayload(msg).build();//此处的txProducerGroup与事务监听器的RocketMQTransactionListener(txProducerGroup

)一致this.rocketMQTemplate.sendMessageInTransaction(test_tx_producer_group,topic,message,null);System.out.println(消息发送成功);}

}生产者监听器

com.rocketmq.spring.transaction;RocketMQTransactionListener(txProducerGroup

public

RocketMQLocalTransactionListener

{private

MapString,RocketMQLocalTransactionState

STATE_MAP

executeLocalTransaction(Message

message,

message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);try

{System.out.println(执行操作1);Thread.sleep(500L);System.out.println(执行操作2);Thread.sleep(500L);STATE_MAP.put(transactionId,RocketMQLocalTransactionState.COMMIT);return

RocketMQLocalTransactionState.COMMIT;}catch

(Exception

e){e.printStackTrace();}STATE_MAP.put(transactionId,RocketMQLocalTransactionState.ROLLBACK);return

RocketMQLocalTransactionState.ROLLBACK;}/***

消息回查**

message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);System.out.println(回查消息-transactionId

transactionId,state

STATE_MAP.get(transactionId));return

}消息生产测试

testSendTransactionMsg(){String

msg

事务消息测试!;this.transactionProducer.sendMsg(test_spring_transaction_topic,msg);System.out.println(发送成功);

}消费者测试

com.rocketmq.spring.transaction;Component

test_spring_transaction_topic,consumeMode

ConsumeMode.CONCURRENTLY,selectorExpression

*,consumerGroup

executeLocalTransaction(Message

message,

message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);try

{System.out.println(执行操作1);Thread.sleep(500L);System.out.println(执行操作2);Thread.sleep(500L);STATE_MAP.put(transactionId,RocketMQLocalTransactionState.COMMIT);return

RocketMQLocalTransactionState.UNKNOWN;}catch

(Exception

e){e.printStackTrace();}STATE_MAP.put(transactionId,RocketMQLocalTransactionState.ROLLBACK);return

RocketMQLocalTransactionState.ROLLBACK;