架构实战篇：Spring Boot 集成 RocketMQ

Posted IT实战联盟

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了架构实战篇：Spring Boot 集成 RocketMQ相关的知识，希望对你有一定的参考价值。

Apache RocketMQ作为阿里开源的一款高性能、高吞吐量的分布式消息中间件。本文快速入门,RocketMQ消息系统的安装部署,发送,和接收消息,监控消息，的详细说明。

环境需要

64位操作系统，建议使用Linux / Unix /

  • CentOs7.3

  • 64bit JDK 1.8+

  • Maven 3.2.x

  • Git 1.8.3.1

环境安装

请参考我的另一篇文章

搭建 Apache RocketMQ 单机环境

http://www.ymq.io/2018/02/01/RocketMQ-install

新加项目

新建一个 maven 项目，这里就不详细操作了，大家都会的

GitHub 源码：https://github.com/souyunku/spring-boot-examples/tree/master/spring-boot-rocketmq

添加依赖

在POM 中添加如下依赖

 
   
   
 
    
  
    
    
  
  1. <!-- RocketMq客户端相关依赖 -->
    2. 
  
    
    
  
  2. <dependency>
    3. 
  
    
    
  
  3.     <groupId>org.apache.rocketmq</groupId>
    4. 
  
    
    
  
  4.     <artifactId>rocketmq-client</artifactId>
    5. 
  
    
    
  
  5.     <version>4.1.0-incubating</version>
    6. 
  
    
    
  
  6. </dependency>
    7. 
  
    
    
  
    8. 
  
    
    
  
  8. <dependency>
    9. 
  
    
    
  
  9.     <groupId>org.apache.rocketmq</groupId>
    10. 
  
    
    
  
  10.     <artifactId>rocketmq-common</artifactId>
    11. 
  
    
    
  
  11.     <version>4.1.0-incubating</version>
    12. 
  
    
    
  
  12. </dependency>
    13.

配置文件

在配置文件 application.properties 添加一下内容

 
   
   
 
    
  
    
    
  
  1. # 消费者的组名
    2. 
  
    
    
  
  2. apache.rocketmq.consumer.PushConsumer=PushConsumer
    3. 
  
    
    
  
    4. 
  
    
    
  
  4. # 生产者的组名
    5. 
  
    
    
  
  5. apache.rocketmq.producer.producerGroup=Producer
    6. 
  
    
    
  
    7. 
  
    
    
  
  7. 
  
    
    
  
  8. apache.rocketmq.namesrvAddr=192.168.252.121:9876
    9.

消息生产者

 
   
   
 
    
  
    
    
  
  1. @Component
    2. 
  
    
    
  
  2. public class Producer {
    3. 
  
    
    
  
    4. 
  
    
    
  
  4.     /**
    5. 
  
    
    
  
  5.      * 生产者的组名
    6. 
  
    
    
  
  6.      */
    7. 
  
    
    
  
  7.     @Value("${apache.rocketmq.producer.producerGroup}")
    8. 
  
    
    
  
  8.     private String producerGroup;
    9. 
  
    
    
  
    10. 
  
    
    
  
  10.     /**
    11. 
  
    
    
  
  11. 
  
    
    
  
  12.      */
    13. 
  
    
    
  
  13.     @Value("${apache.rocketmq.namesrvAddr}")
    14. 
  
    
    
  
  14.     private String namesrvAddr;
    15. 
  
    
    
  
    16. 
  
    
    
  
  16.     @PostConstruct
    17. 
  
    
    
  
  17.     public void defaultMQProducer() {
    18. 
  
    
    
  
    19. 
  
    
    
  
  19.         //生产者的组名
    20. 
  
    
    
  
  20.         DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
    21. 
  
    
    
  
    22. 
  
    
    
  
  22. 
  
    
    
  
  23.         producer.setNamesrvAddr(namesrvAddr);
    24. 
  
    
    
  
    25. 
  
    
    
  
  25.         try {
    26. 
  
    
    
  
    27. 
  
    
    
  
  27.             /**
    28. 
  
    
    
  
  28.              * Producer对象在使用之前必须要调用start初始化，初始化一次即可
    29. 
  
    
    
  
  29.              * 注意：切记不可以在每次发送消息时，都调用start方法
    30. 
  
    
    
  
  30.              */
    31. 
  
    
    
  
  31.             producer.start();
    32. 
  
    
    
  
    33. 
  
    
    
  
  33.             for (int i = 0; i < 100; i++) {
    34. 
  
    
    
  
    35. 
  
    
    
  
  35.                 String messageBody = "我是消息内容:" + i;
    36. 
  
    
    
  
    37. 
  
    
    
  
  37.                 String message = new String(messageBody.getBytes(), "utf-8");
    38. 
  
    
    
  
    39. 
  
    
    
  
  39.                 //构建消息
    40. 
  
    
    
  
  40.                 Message msg = new Message("PushTopic" /* PushTopic */, "push"/* Tag  */, "key_" + i /* Keys */, message.getBytes());
    41. 
  
    
    
  
    42. 
  
    
    
  
  42.                 //发送消息
    43. 
  
    
    
  
  43.                 SendResult result = producer.send(msg);
    44. 
  
    
    
  
    45. 
  
    
    
  
  45.                 System.out.println("发送响应：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());
    46. 
  
    
    
  
    47. 
  
    
    
  
  47.             }
    48. 
  
    
    
  
    49. 
  
    
    
  
  49.         } catch (Exception e) {
    50. 
  
    
    
  
  50.             e.printStackTrace();
    51. 
  
    
    
  
  51.         } finally {
    52. 
  
    
    
  
  52.             producer.shutdown();
    53. 
  
    
    
  
  53.         }
    54. 
  
    
    
  
    55. 
  
    
    
  
  55.     }
    56. 
  
    
    
  
  56. }
    57.

消息消费者

 
   
   
 
    
  
    
    
  
  1. @Component
    2. 
  
    
    
  
  2. public class Consumer {
    3. 
  
    
    
  
    4. 
  
    
    
  
  4.     /**
    5. 
  
    
    
  
  5.      * 消费者的组名
    6. 
  
    
    
  
  6.      */
    7. 
  
    
    
  
  7.     @Value("${apache.rocketmq.consumer.PushConsumer}")
    8. 
  
    
    
  
  8.     private String consumerGroup;
    9. 
  
    
    
  
    10. 
  
    
    
  
  10.     /**
    11. 
  
    
    
  
  11. 
  
    
    
  
  12.      */
    13. 
  
    
    
  
  13.     @Value("${apache.rocketmq.namesrvAddr}")
    14. 
  
    
    
  
  14.     private String namesrvAddr;
    15. 
  
    
    
  
    16. 
  
    
    
  
  16.     @PostConstruct
    17. 
  
    
    
  
  17.     public void defaultMQPushConsumer() {
    18. 
  
    
    
  
    19. 
  
    
    
  
  19.         //消费者的组名
    20. 
  
    
    
  
  20.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
    21. 
  
    
    
  
    22. 
  
    
    
  
  22. 
  
    
    
  
  23.         consumer.setNamesrvAddr(namesrvAddr);
    24. 
  
    
    
  
  24.         try {
    25. 
  
    
    
  
  25.             //订阅PushTopic下Tag为push的消息
    26. 
  
    
    
  
  26.             consumer.subscribe("PushTopic", "push");
    27. 
  
    
    
  
    28. 
  
    
    
  
  28.             //设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
    29. 
  
    
    
  
  29.             //如果非第一次启动，那么按照上次消费的位置继续消费
    30. 
  
    
    
  
  30.             consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    31. 
  
    
    
  
  31.             consumer.registerMessageListener(new MessageListenerConcurrently() {
    32. 
  
    
    
  
    33. 
  
    
    
  
  33.                 @Override
    34. 
  
    
    
  
  34.                 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
    35. 
  
    
    
  
  35.                     try {
    36. 
  
    
    
  
  36.                         for (MessageExt messageExt : list) {
    37. 
  
    
    
  
    38. 
  
    
    
  
  38.                             System.out.println("messageExt: " + messageExt);//输出消息内容
    39. 
  
    
    
  
    40. 
  
    
    
  
  40.                             String messageBody = new String(messageExt.getBody(), "utf-8");
    41. 
  
    
    
  
    42. 
  
    
    
  
  42.                             System.out.println("消费响应：Msg: " + messageExt.getMsgId() + ",msgBody: " + messageBody);//输出消息内容
    43. 
  
    
    
  
    44. 
  
    
    
  
  44.                         }
    45. 
  
    
    
  
  45.                     } catch (Exception e) {
    46. 
  
    
    
  
  46.                         e.printStackTrace();
    47. 
  
    
    
  
  47.                         return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
    48. 
  
    
    
  
  48.                     }
    49. 
  
    
    
  
  49.                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
    50. 
  
    
    
  
  50.                 }
    51. 
  
    
    
  
    52. 
  
    
    
  
    53. 
  
    
    
  
  53.             });
    54. 
  
    
    
  
  54.             consumer.start();
    55. 
  
    
    
  
  55.         } catch (Exception e) {
    56. 
  
    
    
  
  56.             e.printStackTrace();
    57. 
  
    
    
  
  57.         }
    58. 
  
    
    
  
  58.     }
    59. 
  
    
    
  
    60. 
  
    
    
  
  60. }
    61.

启动服务

 
   
   
 
    
  
    
    
  
  1. @SpringBootApplication
    2. 
  
    
    
  
  2. public class SpringBootRocketmqApplication {
    3. 
  
    
    
  
    4. 
  
    
    
  
  4.     public static void main(String[] args) {
    5. 
  
    
    
  
  5.         SpringApplication.run(SpringBootRocketmqApplication.class, args);
    6. 
  
    
    
  
  6.     }
    7. 
  
    
    
  
  7. }
    8.

控制台会有响应

 
   
   
 
    
  
    
    
  
  1. 发送响应：MsgId:0AFF015E556818B4AAC208A0504F0063，发送状态:SEND_OK
    2. 
  
    
    
  
    3. 
  
    
    
  
  3. messageExt: MessageExt [queueId=0, storeSize=195, queueOffset=113824, sysFlag=0, bornTimestamp=1517559124047, bornHost=/192.168.252.1:62165, storeTimestamp=1517559135052, storeHost=/192.168.252.121:10911, msgId=C0A8FC7900002A9F00000000056F499C, commitLogOffset=91179420, bodyCRC=1687852546, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=PushTopic, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=113825, KEYS=key_99, CONSUME_START_TIME=1517559124049, UNIQ_KEY=0AFF015E556818B4AAC208A0504F0063, WAIT=true, TAGS=push}, body=21]]
    4. 
  
    
    
  
    5. 
  
    
    
  
  5. 消费响应：Msg: 0AFF015E556818B4AAC208A0504F0063,msgBody: 我是消息内容:99
    6. 
  
    
    
  
    7. 
  
    
    
  
  7. ...
    8.

监控服务

RocketMQ web界面监控RocketMQ-Console-Ng部署

https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console

下载并且编译

下载并且 maven 编译

 
   
   
 
    
  
    
    
  
  1. git clone https://github.com/apache/rocketmq-externals.git
    2. 
  
    
    
  
  2. rocketmq-externals/rocketmq-console/
    3. 
  
    
    
  
  3. mvn clean package -Dmaven.test.skip=true
    4.

启动监控服务

 
   
   
 
    
  
    
    
  
  1. nohup java -jar target/rocketmq-console-ng-1.0.0.jar --rocketmq.config.namesrvAddr=127.0.0.1:9876
    2.

访问监控服务

架构实战篇（九）：Spring Boot 集成 RocketMQ架构实战篇（九）：Spring Boot 集成 RocketMQ架构实战篇（九）：Spring Boot 集成 RocketMQ架构实战篇（九）：Spring Boot 集成 RocketMQ架构实战篇（九）：Spring Boot 集成 RocketMQ

GitHub 源码：https://github.com/souyunku/spring-boot-examples/tree/master/spring-boot-rocketmq

Gitee 源码：https://gitee.com/souyunku/spring-boot-examples/tree/master/spring-boot-rabbitmq

关注我们

合作伙伴

  • 出处：http://www.ymq.io/2018/02/02/spring-boot-rocketmq-example

  • 版权归作者所有，转载请注明出处

以上是关于架构实战篇：Spring Boot 集成 RocketMQ的主要内容，如果未能解决你的问题，请参考以下文章

架构实战篇：Spring Boot 集成企业级搜索引擎 SolrCloud

架构实战篇（十三）：Spring Boot Logback 邮件通知

Spring Boot Serverless 实战系列“架构篇” | 光速入门函数计算

Spring Boot Serverless 实战系列“架构篇” 首发 | 光速入门函数计算

Spring Boot Serverless 实战系列“部署篇” | Mall 应用

Spring Boot 揭秘与实战 数据缓存篇 - 快速入门