Rocketmq-spring入门与实践

Rocketmq-spring入门与实践

1. 创建资源

开始实验之前，您需要先创建ECS实例资源。

1. 在实验室页面，单击创建资源。
   
2. （可选）在实验室页面左侧导航栏中，单击云产品资源列表，可查看本次实验资源相关信息（例如IP地址、用户信息等）。
   

说明 资源创建过程需要1~3分钟。

2. 启动名称服务器和代理

本步骤指导您如何在已搭建好RocketMQ实例中启动名称服务器和代理。

1. 执行如下命令，切换目录至rocketmq-4.9.3/bin下。
   

cd /root/rocketmq-4.9.3/bin

2. 执行如下命令，启动服务器。
   

说明 按CTRL+C可结束当前进程。

nohup sh mqnamesrv &

tail -f ~/logs/rocketmqlogs/namesrv.log

若启动代理失败，解决方案如下。

1. 执行如下命令，查看 mqnamesrv 进程。
   

ps -ef |grep mqnamesrv

2. 执行如下命令，清除掉该进程（该命令进程编号以上图示例）。
   

kill -9 6987

3. 执行如下命令，清除Java进程。
   

sh mqshutdown namesrv
sh mqshutdown broker

3. 执行如下命令，启动代理。
   

说明 按CTRL+C可结束当前进程。

nohup sh mqbroker -n localhost:9876 &

tail -f ~/logs/rocketmqlogs/broker.log

4.执行如下命令，创建topic。

cd /root/rocketmq-4.9.3/bin
./mqadmin updateTopic -n localhost:9876 -b localhost:10911 -t sendDataType
./mqadmin updateTopic -n localhost:9876 -b localhost:10911 -t sendOperationType
./mqadmin updateTopic -n localhost:9876 -b localhost:10911 -t syncSendOrderly
./mqadmin updateTopic -n localhost:9876 -b localhost:10911 -t sendAndReceive
./mqadmin updateTopic -n localhost:9876 -b localhost:10911 -t broadcastConsumer
./mqadmin updateTopic -n localhost:9876 -b localhost:10911 -t transactionMessage

3. 启动 java 工程

本步骤指导您如何启动java工程。

1. 执行如下命令，启动Java工程。
   

说明 首次运行可能要花费3分钟左右的编译时间。

cd /root/rocketmq-handson-apply
mvn install -Dmaven.test.skip
export local=$RANDOM
java -jar  rocketmq-spring/target/rocketmq-spring-1.0.0.jar

启动成功，返回信息如下。

2. 单击当前页面右上角，再开启一个会话
   

4. 访问并测试执行效果

本步骤指导您如何进行测试，并查看其相应效果。

1. 消息类型测试。

1. 发送对象。
   

执行如下命令，向刚才启动的程序发送一条简单的请求，以观察响应。

wget -q -O- "http://127.0.0.1:28082/messageSend/sendObject?id=1&name=xiaoming&action=go2&ags=3"

再观察会话1中的日志，会观察到如下的输出。

2. 发送message。
   

执行如下命令，向刚才启动的程序发送一条简单的请求，以观察响应。

wget -q -O- "http://127.0.0.1:28082/messageSend/sendMessage?id=2&name=xiaoming&action=des&ags=3"

再观察会话1中的日志，会观察到如下的输出。

3. 发送List。
   

执行如下命令，向刚才启动的程序发送一条简单的请求，以观察响应。

wget -q -O- "http://127.0.0.1:28082/messageSend/sendCollection?id=2&name=xiaoming&action=des&ags=3"

再观察会话1中的日志，会观察到如下的输出。

2. 测试不同发送行为。

1. 同步发送。
   

执行如下命令，向刚才启动的程序发送一条简单的请求，以观察响应。

wget -q -O- "http://127.0.0.1:28082/messageSend/syncSend?id=2&name=xiaoming&action=des&ags=3"

再观察会话1中的日志，会观察到如下的输出。

2. 异步发送。
   

执行如下命令，向刚才启动的程序发送一条简单的请求，以观察响应。

wget -q -O- "http://127.0.0.1:28082/messageSend/asyncSend?id=2&name=xiaoming&action=des&ags=3"

再观察会话1中的日志，会观察到如下的输出。

3. OneWay发送。
   

执行如下命令， 向刚才启动的程序发送一条简单的请求，以观察响应。

wget -q -O- "http://127.0.0.1:28082/messageSend/oneWaySend?id=2&name=xiaoming&action=des&ags=3"

再观察会话1中的日志，会观察到如下的输出。

4. 发送事务消息。
   

执行如下命令， 向刚才启动的程序发送一条简单的请求，以观察响应。

wget -q -O- "http://127.0.0.1:28082/messageSend/transactionMessage?id=2&name=xiaoming&action=des&arg=1"
wget -q -O- "http://127.0.0.1:28082/messageSend/transactionMessage?id=2&name=xiaoming&action=des&arg=2"
wget -q -O- "http://127.0.0.1:28082/messageSend/transactionMessage?id=2&name=xiaoming&action=des&arg=3"

再观察会话1中的日志，会观察到如下的输出。

5. 发送同步RPC消息。
   

执行如下命令，向刚才启动的程序发送一条简单的请求，以观察响应。

wget -q -O- "http://127.0.0.1:28082/messageSend/sendAndReceive?id=2&name=xiaoming&action=des&ags=3"

预期返回结果如下。

再观察会话1中的日志，会观察到如下的输出。

5. 细节点

本步骤将讲解rocketmq-spring的tag设置与原生不同以及关于 Message 对象的 header。

1. rocketmq-spring的tag设置与原生不同。
   

现在把topic与tag合成了destination。destination=topic or topic":"tag。实例请看MessageSendController第152行。

2. 关于 Message 对象的 header。
   

用于为消息提供附加信息，同时降低了消息体反序列化带来的额外开销。

1. RocketMQ默认请求头。
   

6. 生产者

本步骤将讲解生产者的创捷和各属性信息。

1. 创建生产者。

1. 默认生产者创建如下。
   

2. 自定义生产者创建如下。
   

3. 继承RocketMQTemplate对象。
   
4. 使用ExtRocketMQTemplateConfiguration 注解。
   

5. 配置属性信息。
   
6. 声明自定义消费者的全局变量，并使用@Autowired注解。
   
7. 消息生产。
   
8. 发送形参说明如下。
   

2. 消息类型解读。
   

说明 三种消息都是使用convert进行数据序列化。本质上一样，不建议使用convertAndSend方式。

3. 生产方式。
   

4. 事务消息。

1. 实现 RocketMQLocalTransactionListener 接口。
   
2. 配置 @RocketMQTransactionListener 注解。
   
3. rocketMQTemplateBeanName 拦截的 RocketMQTemplate 的事务，值对应ExtRocketMQTemplateConfiguration.value。
   

5. 实现接口方法。
   

实验链接：https://developer.aliyun.com/adc/scenario/470077b9392a482b91fa1a2bf4c423fe