Rocketmq源码--3--整合springboot

分手后的思念是犯贱 2022-10-16 09:54 116阅读 0赞

# `Rocketmq`源码–`3`–整合`springboot` #

### 文章目录 ###

*  \`Rocketmq\`源码--\`3\`--整合\`springboot\`
 *   *  1 准备工作
     *  2 发送一条同步消息并消费
     *   *  2.1 向\`Rocketmq\`发送消息
         *  2.2 消费\`Rocketmq\`的消息
     *  3 发送一条异步消息并消费
     *  4 发送一条单向消息并消费
     *  5 发送一条\`json\`消息并消费
     *  6 发送一条延时消息并消费
     *  7 消费重试

## 1 准备工作 ##

这里我们创建一个父工程，他下面有两个子工程，分别是消费者和生产者工程。

在父工程中导入依赖

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.3.RELEASE</version>
    </parent>
    
    <dependencies>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.2</version>
        </dependency>
    </dependencies>

各个子工程的配置文件除了`spirngboot`的基本配置，还需要加入`Rocketmq`的配置

rocketmq:
      nameServer: 192.168.43.33:9876
      #消费者工程中不需要加入下面的配置
      producer:
        group: demo-producer-group

## 2 发送一条同步消息并消费 ##

### 2.1 向`Rocketmq`发送消息 ###

我们这里测试向`Rocketmq`发送消息

@SpringBootTest
    @RunWith(SpringRunner.class)
    public class ProducerTest { 
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        /** * 发送同步消息 */
        @Test
        public void sendSyncMsg() { 
            rocketMQTemplate.syncSend("my-topic","第一条同步消息");
            System.out.println("发送成功");
        }
    }

控制台打印`发送成功`之后，我们就去`Rocketmq`的控制台页面

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center 1]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center 2]

### 2.2 消费`Rocketmq`的消息 ###

创建一个类，该类用来接收`Rocketmq`发送的消息

/** * topic 监听的主题 * consumerGroup 消费组，相同消费组的消费者共同消费该主题的消息，它们组成一个集群 * demo-producer-group 是我们在生产者的配置文件中设置的 */
    @Component
    @RocketMQMessageListener(topic = "my-topic", consumerGroup = "demo-producer-group")
    public class ConsumerSimple implements RocketMQListener<String> { 
    
        /** * 消息接收的方法 * * @param message */
        @Override
        public void onMessage(String message) { 
    
            System.out.println(message);
        }
    }

观察控制台是否打印`第一条同步消息`，然后看`Rocketmq`的控制台

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center 3]

## 3 发送一条异步消息并消费 ##

向`Rocketmq`发送消息

@SpringBootTest
    @RunWith(SpringRunner.class)
    public class ProducerTest { 
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
         /** * 发送异步消息 * 消息发送完成后，不管是否发送成功与否，都直接返回， * 发送结果会新建一个线程通过回调函数通知 */
        @Test
        public void sendAsyncMsg() throws InterruptedException { 
    
            rocketMQTemplate.asyncSend("my-topic","第1条异步消息",new SendCallback(){ 
    
                @Override
                public void onSuccess(SendResult sendResult) { 
                    System.out.println("发送成功");
                }
    
                @Override
                public void onException(Throwable e) { 
                    System.out.println("发送失败");
                }
            });
            //异步消息，为跟踪回调线程这里加入延迟，
            // 如果不加，会导致主线程结束，消息直接发送失败
            Thread.sleep(3000);
        }
    }

控制台打印`发送成功`之后，说明我们消息已经异步发送成功了，启动消费者，控制台打印出`第1条异步消息`，消费成功

## 4 发送一条单向消息并消费 ##

向`Rocketmq`发送消息

@SpringBootTest
    @RunWith(SpringRunner.class)
    public class ProducerTest { 
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
         /** * 发送单向消息 * 消息发送完成后，不管是否发送成功与否，都直接返回，不等待broker 服务器的结果 */
        @Test
        public void sendOneWayMsg(){ 
            rocketMQTemplate.sendOneWay("my-topic","第1条单向消息");
        }
    }

先启动消费者，然后发送一条单向消息，消费者控制台打印出`第1条单向消息`，消费成功

## 5 发送一条`json`消息并消费 ##

@Data
    @NoArgsConstructor
    @AllArgsConstructor
    @ToString
    public class OrderExt implements Serializable { 
    
        private String id;
        private Date createTime;
        private Long money;
        private String title;
    }

向`Rocketmq`发送消息

@SpringBootTest
    @RunWith(SpringRunner.class)
    public class ProducerTest { 
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        /** * 发送json格式的消息 */
        @Test
        public void sendJsonMsg(){ 
            OrderExt orderExt = new OrderExt(
                    "12",
                    new Date(),
                    12l,
                    "测试"
            );
            //rocketMQTemplate.syncSend("my-topic",orderExt);
            //使用message构造器构造消息
            Message<OrderExt> build = MessageBuilder.withPayload(orderExt).build();
            rocketMQTemplate.syncSend("my-topic",build);
    
        }
    }

先启动消费者，然后发送一条同步消息，消费者控制台打印出`orderExt`对象的信息，消费成功

## 6 发送一条延时消息并消费 ##

延迟消息也叫做定时消息，比如在电商项目的交易系统中，当用户下单之后超过一段时间之后仍然没有支付，此时就需要将该订单关闭。要实现该功能的话，可以在用户创建订单时就发送一条包含订单内容的延迟消息，该消息在 一段时间之后投递给消息消费者，当消息消费者接收到该消息后，判断该订单的支付状态，如果处于未支付状态， 则将该订单关闭。

`RocketMQ`的延迟消息实现非常简单，只需要发送消息前设置延迟的时间，延迟时间存在十八个等级`（1s/5s/10s/30s/1m/2m/3m/4m/5m/6m/7m/8m/9m/10m/20m/30m/1h/2h）`，调用`setDelayTimeLevel()`设置与时间相对应的延迟级别即可。

向`Rocketmq`发送消息

@SpringBootTest
    @RunWith(SpringRunner.class)
    public class ProducerTest { 
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        /** * 发送json格式的延时消息 */
        @Test
        public void sendJsonDelayMsg() throws InterruptedException, RemotingException, MQClientException, MQBrokerException { 
            OrderExt orderExt = new OrderExt(
                    "12567",
                    new Date(),
                    12l,
                    "测试"
            );
            //同步
            //Message<OrderExt> build = MessageBuilder.withPayload(orderExt).build();
            //rocketMQTemplate.syncSend("my-topic",build,3000,5);
            //异步
            org.apache.rocketmq.common.message.Message message = new
                    org.apache.rocketmq.common.message.Message(
                    "my-topic", JSON.toJSONBytes(orderExt));
            //设置延时等级
            message.setDelayTimeLevel(5);
            //通过取得生产者的操作对象来发送消息，既可发同步也可发异步
            rocketMQTemplate.getProducer().send(message, new SendCallback() { 
                @Override
                public void onSuccess(SendResult sendResult) { 
                    System.out.println(sendResult);
                }
    
                @Override
                public void onException(Throwable throwable) { 
                    System.out.println(throwable.getMessage());
                }
            });
            Thread.sleep(3000);
        }
    }

先启动消费者，然后测试方法，消费者控制台打印出`orderExt`对象的信息，消费成功

## 7 消费重试 ##

当消费没有成功，会触发该机制。

`Rocketmq`并不会一直进行消费重试，它最多只会重试`16`次，然后该条消息就进入了死信队列了。

消费重试的时候和延迟消息差不多，它的第一条重试消息从第三个等级开始（也就是消息发送失败`10s`后重新发送），每发送失败一次增加一个等级，一直到最大等级后，`Rocketmq`就会将这条消息放入死信队列。

在实际开发中，我们并不会让`Rocketmq`进行消费重试一直到进入死信队列，我们会监听消息的重试次数，当次数达到了我们规定的之后，我们将这条消息写入数据库中，由人工进行处理。

在消费者中

/** * topic 监听的主题 * consumerGroup 消费组，相同消费组的消费者共同消费该主题的消息，它们组成一个集群 * demo-producer-group 是我们在生产者的配置文件中设置的 * MessageExt相比于string，内容更加丰富，包含了队列和主题的信息 */
    @Component
    @RocketMQMessageListener(topic = "my-topic", consumerGroup = "demo-producer-group")
    public class ConsumerSimple implements RocketMQListener<MessageExt> { 
    
    
        /** * 消息接收的方法 * @param message */
        @Override
        public void onMessage(MessageExt message) { 
            int reconsumeTimes = message.getReconsumeTimes();
            if (reconsumeTimes>2){ 
                //将消息取出保存到数据库的操作
                try { 
                    System.out.println(new String(message.getBody(),"utf-8"));
                } catch (UnsupportedEncodingException e) { 
                    e.printStackTrace();
                }
                return;
            }
            throw new RuntimeException("模拟消费失败");
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center]: /images/20221014/05d454c519814a4189a244a5cd03bdb9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221014/98ea8f7cc55649a3a56255bed9b11bb8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221014/5d59d77d46a04a1da80b5c2f1f1ec7d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMDEyOTgx_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221014/ef15e47430014584a8ca2dd9b2ee5616.png