1、安装 RocketMQ

下载：rocketmq-all-4.8.0-bin-release.zip

1、传入 Linux 服务器

2、解压缩

  1. unzip rocketmq-all-4.8.0-bin-release.zip

3、调整启动参数，

  1. cd rocketmq-all-4.8.0-bin-release/bin

修改默认启动参数，默认启动的最大内存为4G，比较大，修改小一点，否则如果服务器内存不够会启动失败

调整namesrv

  1. vim runserver.sh

调整如下

Spring Cloud Alibaba 05：RocketMQ - 图1

调整broker

  1. vim runbroker.sh

调整如下

Spring Cloud Alibaba 05：RocketMQ - 图2

4、启动namesrv和启动broker

启动navmesrv

  1. nohup sh mqnamesrv &

启动broker,注意ip为公网ip,端口为navmesrv的默认端口9876

  1. nohup ./mqbroker -n localhost:9876 &

5、检查是否启动成功

  1. jps -l

Spring Cloud Alibaba 05：RocketMQ - 图3

也可以查看日志

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

启动成功

6、测试 RocketMQ

消息发送

  1. export NAMESRV_ADDR=localhost:
  2. ./tools.sh org.apache.rocketmq.example.quickstart.Producer

消息接收

  1. ./tools.sh org.apache.rocketmq.example.quickstart.Consumer

7、关闭 RocketMQ

  1. ./mqshutdown broker
  2. ./mqshutdown namesrv

2、安装 RocketMQ 控制台

  1. git clone  https://github.com/apache/rocketmq-externals.git

1、进入到rocketmq-console的配置文件，修改如下：

Spring Cloud Alibaba 05：RocketMQ - 图4

2、打包

  1. mvn clean package -Dmaven.test.skip=true

3、进入 target 启动 jar

  1. java -jar rocketmq-console-ng-2.0.0.jar

打开浏览器访问 localhost:9877，如果报错

Spring Cloud Alibaba 05：RocketMQ - 图5

这是因为我们的 RocketMQ 安装在 Linux 中，控制台在 windows，Linux 需要开放端口才能访问，开放 10909 和 9876 端口

  1. firewall-cmd --zone=public --add-port=10909/tcp --permanent
  2. firewall-cmd --zone=public --add-port=10911/tcp --permanent
  3. firewall-cmd --zone=public --add-port=9876/tcp --permanent
  4. systemctl restart firewalld.service
  5. firewall-cmd --reload

重新启动控制台项目

3、Java 实现消息发送

1、pom.xml 中引入依赖

  1. <dependency>
  2.     <groupId>org.apache.rocketmq</groupId>
  3.     <artifactId>rocketmq-spring-boot-starter</artifactId>
  4.     <version>2.1.1</version>
  5. </dependency>

2、生产消息

  1. package com.godfrey;
  3. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  4. import org.apache.rocketmq.client.producer.SendResult;
  5. import org.apache.rocketmq.common.message.Message;
  7. @SpringBootTest
  8. class ProviderApplicationTests {
  10.     @Test
  11.     @DisplayName("测试RocketMQ消息发送")
  12.     void test3() throws Exception {
  13.         //创建消息生产者
  14.         DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
  15.         //设置NameServer
  16.         producer.setNamesrvAddr("39.106.41.184:9876");
  17.         //启动生产者
  18.         producer.start();
  19.         //构建消息对象
  20.         Message message = new Message("myTopic", "myTag", ("Test MQ").getBytes());
  21.         //发送消息
  22.         SendResult result = producer.send(message, 1000);
  23.         System.out.println(result);
  24.         //关闭生产者
  25.         producer.shutdown();
  26.     }
  27. }

3、直接运行，如果报错 sendDefaultImpl call timeout，可以开放 10911 端口

  1. firewall-cmd --zone=public --add-port=10911/tcp --permanent
  2. systemctl restart firewalld.service
  3. firewall-cmd --reload

打开 RocketMQ 控制台，可查看消息。

4、Java 实现消息消费

  1. package com.godfrey;
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  5. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  6. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  7. import org.apache.rocketmq.client.exception.MQClientException;
  8. import org.apache.rocketmq.common.message.MessageExt;
  10. import java.util.List;
  12. @SpringBootTest
  13. class ProviderApplicationTests {
  15.     @Test
  16.     @DisplayName("测试RocketMQ消息接收")
  17.     void test4() throws MQClientException {
  18.         //创建消息消费者
  19.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer-group");
  20.         //设置NameServer
  21.         consumer.setNamesrvAddr("39.106.41.184:9876");
  22.         //指定订阅的主题和标签
  23.         consumer.subscribe("myTopic", "*");
  24.         //回调函数
  25.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  26.             @Override
  27.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
  28.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  29.             }
  30.         });
  31.         //启动消费者
  32.         consumer.start();
  33.     }
  34. }

5、Spring Boot 整合 RocketMQ

provider

1、pom.xml

  1. <dependency>
  2.     <groupId>org.apache.rocketmq</groupId>
  3.     <artifactId>rocketmq-spring-boot-starter</artifactId>
  4.     <version>2.1.1</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>org.apache.rocketmq</groupId>
  8.     <artifactId>rocketmq-client</artifactId>
  9.     <version>4.8.0</version>
  10. </dependency>

2、application.yml

  1. rocketmq:
  2.   name-server: 39.106.41.184:9876
  3.   producer:
  4.     group: myprovider

3、Order

  1. package com.godfrey.entity;
  3. import java.io.Serializable;
  4. import java.util.Date;
  6. /**
  7.  * @author godfrey
  8.  * @since 2020-12-27
  9.  */
  10. public class Order implements Serializable {
  11.     private static final long serialVersionUID = -5397628182599822017L;
  12.     private Integer id;
  13.     private String buyerName;
  14.     private String buyerTel;
  15.     private String address;
  16.     private Date createDate;
  18.     public Order() {
  19.     }
  21.     public Order(Integer id, String buyerName, String buyerTel, String address, Date createDate) {
  22.         this.id = id;
  23.         this.buyerName = buyerName;
  24.         this.buyerTel = buyerTel;
  25.         this.address = address;
  26.         this.createDate = createDate;
  27.     }
  29.     public Integer getId() {
  30.         return id;
  31.     }
  33.     public void setId(Integer id) {
  34.         this.id = id;
  35.     }
  37.     public String getBuyerName() {
  38.         return buyerName;
  39.     }
  41.     public void setBuyerName(String buyerName) {
  42.         this.buyerName = buyerName;
  43.     }
  45.     public String getBuyerTel() {
  46.         return buyerTel;
  47.     }
  49.     public void setBuyerTel(String buyerTel) {
  50.         this.buyerTel = buyerTel;
  51.     }
  53.     public String getAddress() {
  54.         return address;
  55.     }
  57.     public void setAddress(String address) {
  58.         this.address = address;
  59.     }
  61.     public Date getCreateDate() {
  62.         return createDate;
  63.     }
  65.     public void setCreateDate(Date createDate) {
  66.         this.createDate = createDate;
  67.     }
  69.     @Override
  70.     public String toString() {
  71.         return "Order{" +
  72.                 "id=" + id +
  73.                 ", buyerName='" + buyerName + '\'' +
  74.                 ", buyerTel='" + buyerTel + '\'' +
  75.                 ", address='" + address + '\'' +
  76.                 ", createDate=" + createDate +
  77.                 '}';
  78.     }
  79. }

4、Controller

  1. private RocketMQTemplate rocketMQTemplate;
  3. @Autowired
  4. public ProviderController(RocketMQTemplate rocketMQTemplate) {
  5.     this.rocketMQTemplate = rocketMQTemplate;
  6. }
  8. @GetMapping("/create")
  9. public Order create(){
  10.     Order order = new Order(
  11.         1,
  12.         "张三",
  13.         "123123",
  14.         "软件园",
  15.         new Date()
  16.     );
  17.     this.rocketMQTemplate.convertAndSend("myTopic",order);
  18.     return order;
  19. }

consumer

1、pom.xml

  1. <dependency>
  2.     <groupId>org.apache.rocketmq</groupId>
  3.     <artifactId>rocketmq-spring-boot-starter</artifactId>
  4.     <version>2.1.1</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>org.apache.rocketmq</groupId>
  8.     <artifactId>rocketmq-client</artifactId>
  9.     <version>4.8.0</version>
  10. </dependency>

2、application.yml

  1. rocketmq:
  2.   name-server: 39.106.41.184:9876

3、Order

  1. package com.godfrey.entity;
  3. import java.io.Serializable;
  4. import java.util.Date;
  6. /**
  7.  * @author godfrey
  8.  * @since 2020-12-27
  9.  */
  10. public class Order implements Serializable {
  11.     private static final long serialVersionUID = -5397628182599822017L;
  12.     private Integer id;
  13.     private String buyerName;
  14.     private String buyerTel;
  15.     private String address;
  16.     private Date createDate;
  18.     public Order() {
  19.     }
  21.     public Order(Integer id, String buyerName, String buyerTel, String address, Date createDate) {
  22.         this.id = id;
  23.         this.buyerName = buyerName;
  24.         this.buyerTel = buyerTel;
  25.         this.address = address;
  26.         this.createDate = createDate;
  27.     }
  29.     public Integer getId() {
  30.         return id;
  31.     }
  33.     public void setId(Integer id) {
  34.         this.id = id;
  35.     }
  37.     public String getBuyerName() {
  38.         return buyerName;
  39.     }
  41.     public void setBuyerName(String buyerName) {
  42.         this.buyerName = buyerName;
  43.     }
  45.     public String getBuyerTel() {
  46.         return buyerTel;
  47.     }
  49.     public void setBuyerTel(String buyerTel) {
  50.         this.buyerTel = buyerTel;
  51.     }
  53.     public String getAddress() {
  54.         return address;
  55.     }
  57.     public void setAddress(String address) {
  58.         this.address = address;
  59.     }
  61.     public Date getCreateDate() {
  62.         return createDate;
  63.     }
  65.     public void setCreateDate(Date createDate) {
  66.         this.createDate = createDate;
  67.     }
  70.     @Override
  71.     public String toString() {
  72.         return "Order{" +
  73.                 "id=" + id +
  74.                 ", buyerName='" + buyerName + '\'' +
  75.                 ", buyerTel='" + buyerTel + '\'' +
  76.                 ", address='" + address + '\'' +
  77.                 ", createDate=" + createDate +
  78.                 '}';
  79.     }
  80. }

4、Service

  1. package com.godfrey.service;
  3. import com.godfrey.entity.Order;
  4. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  5. import org.apache.rocketmq.spring.core.RocketMQListener;
  6. import org.slf4j.Logger;
  7. import org.slf4j.LoggerFactory;
  8. import org.springframework.stereotype.Service;
  10. /**
  11.  * @author godfrey
  12.  * @since 2020-12-27
  13.  */
  14. @Service
  15. @RocketMQMessageListener(consumerGroup = "myConsumer", topic = "myTopic")
  16. public class SmsService implements RocketMQListener<Order> {
  18.     private static final Logger log = LoggerFactory.getLogger(SmsService.class);
  20.     @Override
  21.     public void onMessage(Order order) {
  22.         log.info("新订单{},发短信通知用户", order);
  23.     }
  24. }