0
点赞
收藏
分享

微信扫一扫

Spring Cloud 集成阿里rocketMq 发送延时/定时/普通消息 解决消费轨迹未消费问题

香小蕉 2021-09-19 阅读 90
spring cloud

spring cloud stream 介绍（照搬）

  • Binding 在消息中间件与应用程序提供的 Provider 和 Consumer 之间提供了一个桥梁，实现了开发者只需使用应用程序的 Provider 或 Consumer 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。

版本选择

一. pom.xml 配置

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.2</version>
    </parent>
    <groupId>org.example</groupId>
    <artifactId>spring-alibaba-cloud-rocketmq-studytest</artifactId>
    <version>1.0-SNAPSHOT</version>
    <description>阿里巴巴cloud-rocketmq学习</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencyManagement>
        <dependencies>
            <!-- spring cloud 版本依赖-->
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>2020.0.0</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <!-- spring alibaba cloud 版本依赖-->
            <dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-alibaba-dependencies</artifactId>
                <version>2021.1</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <!-- rocketmq 依赖-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
        </dependency>
        <!-- spring boot web依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>
</project>

二. 自定义消息channel与rocketMq配置

  • 1. 自定义普通消息

  • 普通消息YML配置
spring:
  cloud:
    stream:
      # 阿里rocketMq配置 topic 与 group 均以 实例id% 为前缀配置 如实例id为 MQ_INST_XXXX_XXX 则group或topic 配置 MQ_INST_XXXX_XXX%grouID
      rocketmq:
        binder:
          # 【若为阿里云购买服务，则为控制台的对外或对内实例地址】【若自己搭建的服务，为自定义rocketmq服务地址127.0.0.1:9876】
          name-server: http://MQ_INST_XXXX_XXX.DD.FFF.aliyuncs.com:80
          # 阿里access-key 【购买阿里服务 控制台获取填写 若为自搭服务可不填】
          access-key: AAAAAAAAAAA
          # 阿里secret-key 【购买阿里服务 控制台获取填写 若为自搭服务可不填】
          secret-key: BBBBBBBBBBBBBB
        # rocketMq 自定义消息通道配置
        bindings:
          # 阿里rocketMq binder 生产者配置
          ### 普通生产消息通道
          customized_output_channel: {producer.group: MQ_INST_XXXX_XXX%GID_QIGUANBANG_DEV}
          # 阿里rocketMq binder 消费者配置
          ### 普通消息订阅通道
          customized_input_channel: {consumer.tags: test_consumer_tag}

      # stream binder 公共配置
      bindings:
        # spring cloud stream binder 生产者配置
        ### 普通消息通道
        customized_output_channel: {destination: MQ_INST_XXXX_XXX%mg_common_topic, content-type: application/json}
        # spring cloud stream binder 消费者配置
        ### 普通消息订阅通道
        customized_input_channel: {destination: MQ_INST_XXXX_XXX%mg_common_topic, group: MQ_INST_XXXX_XXX%GID_QIGUANBANG_DEV, content-type: application/json}
  • 自定义channel接口
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;

@Slf4j
@EnableBinding({Source.class, Sink.class})
@SpringBootApplication
public class RocketApplication {
    public static void main(String[] args) {
        SpringApplication.run(RocketApplication.class, args);
        log.debug("==========rocketMq服务启动成功！==========");
    }
}

@Component
@EnableBinding(Source.class)
public class RocketMqService {
  • source
package org.springframework.cloud.stream.messaging;

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;

/**
 * Bindable interface with one output channel.
 *
 * @author Dave Syer
 * @author Marius Bogoevici
 * @see org.springframework.cloud.stream.annotation.EnableBinding
 */
public interface Source {

    /**
     * Name of the output channel.
     */
    String OUTPUT = "output";

    /**
     * @return output channel
     */
    @Output(Source.OUTPUT)
    MessageChannel output();

}
  • sink
package org.springframework.cloud.stream.messaging;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;

/**
 * Bindable interface with one input channel.
 *
 * @author Dave Syer
 * @author Marius Bogoevici
 * @see org.springframework.cloud.stream.annotation.EnableBinding
 */
public interface Sink {

    /**
     * Input channel name.
     */
    String INPUT = "input";

    /**
     * @return input channel.
     */
    @Input(Sink.INPUT)
    SubscribableChannel input();

}
  • 自定义消息生产通道接口
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;

public interface OutputChannel {
    // 普通消息生产通道 对应yml自定义节点名称
    String NORMAL_PRODUCER_CHANNEL = "customized_output_channel";

    @Output(NORMAL_PRODUCER_CHANNEL)
    MessageChannel NormalOutput();
}
  • 自定义消息订阅通道接口
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;

public interface InputChannel {
    // 普通消息订阅通道 对应yml自定义节点名称
    String NORMAL_CONSUMER_CHANNEL = "customized_input_channel";

    @Input(NORMAL_CONSUMER_CHANNEL)
    SubscribableChannel normalConsumerChannel();
}
启动类启用
import com.study.rocketmq.channel.InputChannel;
import com.study.rocketmq.channel.OutputChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;

@Slf4j
@EnableBinding({InputChannel.class, OutputChannel.class})
@SpringBootApplication
public class RocketApplication {
    public static void main(String[] args) {
        SpringApplication.run(RocketApplication.class, args);
        log.debug("==========rocketMq服务启动成功！==========");
    }
}
  • 普通消息发送
// controller
@RestController
@RequestMapping("/msg")
public class TestMsgController {

    @Autowired
    ProducerService producerService;

    @GetMapping("/sendMsg/{msg}")
    public String sendMsg(@PathVariable("msg")String msg){
        producerService.sendNormalMsg(msg, "test_consumer_tag", "testKey");
        return "SUCCESS";
    }
}

import com.study.rocketmq.channel.OutputChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class ProducerService {

    @Autowired
    private OutputChannel outputChannel;
    /**
     * 发送普通消息
     * @param message       消息内容
     * @param ConsumerTag   消费者group标识
     * @param MsgKey        消息key
     * @return
     */
    public boolean sendNormalMsg(String message, String ConsumerTag, String MsgKey){
        // 构建消息
        Message<String> messageBuild = MessageBuilder.withPayload(message)
                .setHeader(MessageConst.PROPERTY_TAGS, ConsumerTag)
                .setHeader(MessageConst.PROPERTY_KEYS, MsgKey)
                .build();
        // 发送消息
        boolean sendResult = outputChannel.NormalOutput().send(messageBuild);
        if (sendResult){
            log.info("普通消息发送成功-ConsumerTag:{}-MsgKey:{}", ConsumerTag, MsgKey);
        }else {
            log.error("普通消息发送失败！：{}",  ConsumerTag, MsgKey);
        }
        return sendResult;
    }
}

import com.study.rocketmq.channel.InputChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
  * @ClassName MessageListener
  * @author lgq
  * @description //消息监听
  * @Date 2021/6/9
  * @Version V1.0
  */
@Slf4j
@Component
public class MessageListener {
    
    // 通过StreamListener监听消息 只允许rocketmq_KEYS = testKey接收
    @StreamListener(value = InputChannel.NORMAL_CONSUMER_CHANNEL, condition = "headers['rocketmq_KEYS'] == 'testKey'")
    public void receivePayMsg(@Payload String payResult) {
        log.debug("接收到普通消息：{}", payResult);
    }
}

后来通过查询资料得知，可能阿里的rocketMq服务版本比较高，ons客户端版本已经到了4.8而spring-cloud-starter-stream-rocketmq所使用的版本才4.4.0，所以我们排除掉它自带的依赖，引入最新的。


<!-- rocketmq 依赖-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
            <!-- 排除自带rocketMq-client依赖【低版本消息无法发送成功】-->
            <exclusions>
                <exclusion>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-client</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-acl</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!-- rocketMq -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.8.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-acl</artifactId>
            <version>4.8.0</version>
        </dependency>

  • 2. 自定义延时/定时消息

    • YML 添加如下配置
             bindings:
             # 阿里rocketMq binder 生产者配置
             ### 延时消息生产 producer.sync 属性需设置为true
             delay_output_channel: {producer.group: MQ_INST_XXXX_XXX%GID_QIGUANBANG_DEV, producer.sync: true}
             # 阿里rocketMq binder 消费者配置
             ### 延时消息订阅
             delay_input_channel: {consumer.tags: test_delay_tag}
         bindings:
           # spring cloud stream binder 生产者配置
           ### 延时消息
           delay_output_channel: {destination: MQ_INST_XXXX_XXX%common-delay-topic, content-type: application/json}
           # spring cloud stream binder 消费者配置
           ### 延时消息订阅
           delay_input_channel: {destination: MQ_INST_XXXX_XXX%common-delay-topic, group: MQ_INST_XXXX_XXX%GID_QIGUANBANG_DEV, content-type: application/json}
    • InputChannel 接口添加如下方法
        // 延时/定时消息订阅通道 对应yml自定义节点名称
   String DELAY_CONSUMER_CHANNEL = "delay_input_channel";
   
   // 延时/定时消息订阅
   @Input(DELAY_CONSUMER_CHANNEL)
   SubscribableChannel delayConsumerChannel();
    • OutputChannel 接口添加如下方法
     // 延时或定时消息生产通道
 String DELAY_PRODUCER_CHANNEL = "delay_output_channel";

 @Output(DELAY_PRODUCER_CHANNEL)
 MessageChannel delayOutput();
    • ProducerService 添加如下方法
         /**
  * 延时消息发送
  * @param message   延时消息体
  * @param delayLevel 延时级别 1~18 (1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 【 1=1s 2=5s 3=10s】)
  * @param ConsumerTag 消费者TAG标识 通过TAG区分消费对象
  * @param MsgKey  消息key 可以通过该字段再次区分
  * @return
  */
 public boolean sendDelayMsg(String message, int delayLevel, String ConsumerTag, String MsgKey){
     // 构建消息
     Message<String> messageBuild = MessageBuilder.withPayload(message)
             .setHeader(MessageConst.PROPERTY_TAGS, ConsumerTag)
             .setHeader(MessageConst.PROPERTY_KEYS, MsgKey)
             .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayLevel)
             .build();
     // 发送消息
     boolean sendResult = outputChannel.delayOutput().send(messageBuild);
     if (sendResult){
         log.info("延时消息发送成功-ConsumerTag:{}-MsgKey:{}", ConsumerTag, MsgKey);
     }else {
         log.error("延时消息发送失败！：{}",  ConsumerTag, MsgKey);
     }
     return sendResult;
 }


 /**
  * https://help.aliyun.com/document_detail/43349.html
  * rocketMq 指定时间消息发送
  * @param message 消息内容
  * @param ConsumerTag 消费者group标识
  * @param MsgKey 消息key
  * @param fixedTime 指定时间戳 指定时间戳必须大于当前时间 否则立即消费 参数可设置40天内的任何时刻（单位毫秒），超过40天消息发送将失败
  * @return
  */
 public boolean sendFixedTimeMsg(String message, String ConsumerTag, String MsgKey, long fixedTime){
     // 构建消息 __STARTDELIVERTIME 为发送定时任务需要的请求头
     Message<String> messageBuild = MessageBuilder.withPayload(message)
             .setHeader(MessageConst.PROPERTY_TAGS, ConsumerTag)
             .setHeader(MessageConst.PROPERTY_KEYS, MsgKey)
             .setHeader("__STARTDELIVERTIME", fixedTime)
             .build();
     // 发送消息
     boolean sendResult = outputChannel.delayOutput().send(messageBuild);
     if (sendResult){
         log.info("定时消息发送成功-ConsumerTag:{}-MsgKey:{}", ConsumerTag, MsgKey);
     }else {
         log.error("定时消息发送失败！：{}",  ConsumerTag, MsgKey);
     }
     return sendResult;
 }
    • MessageListener 监听器添加如下监听
     // 监听定时/延时消息通道，只允许key = delayMsg 通过
 @StreamListener(value = InputChannel.DELAY_CONSUMER_CHANNEL, condition = "headers['rocketmq_KEYS'] == 'delayMsg'")
 public void receiveDelayMsg(@Payload String payResult) {
     log.debug("接收到延时消息：{}", payResult);
 }

 // 监听定时/延时消息通道，只允许key = fixTimeMsg通过
 @StreamListener(value = InputChannel.DELAY_CONSUMER_CHANNEL, condition = "headers['rocketmq_KEYS'] == 'fixTimeMsg'")
 public void receivefixTimeMsg(@Payload String payResult) {
     log.debug("接收到定时消息：{}", payResult);
 }

      @GetMapping("/sendMsg/{msg}")
  public String sendMsg(@PathVariable("msg")String msg){
      producerService.sendDelayMsg(msg, 2,"test_delay_tag", "delayMsg");
      return "SUCCESS";
  }

      @GetMapping("/sendMsg/{msg}")
  public String sendMsg(@PathVariable("msg")String msg) throws ParseException {
      SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
      Calendar calendar = Calendar.getInstance();
      calendar.setTime(simpleDateFormat.parse("2021-06-09 18:18:00"));
      long time = calendar.getTime().getTime();

      producerService.sendFixedTimeMsg(msg, "test_delay_tag", "delayMsg", time);
      return "SUCCESS";
  }

三. application.yml 完整配置

spring:
  application:
    name: rocketmq-server
  cloud:
    stream:
      # 阿里rocketMq配置 topic 与 group 均以 实例id% 为前缀配置 如实例id为 MQ_INST_XXXX_XXX 则group或topic 配置 MQ_INST_XXXX_XXX%grouID
      rocketmq:
        binder:
          # 【若为阿里云购买服务，则为控制台的对外或对内实例地址】【若自己搭建的服务，为自定义rocketmq服务地址127.0.0.1:9876】
          name-server: http://MQ_INST_XXXX_XXX.mq-internet-access.mq-internet.aliyuncs.com:80
          # 阿里access-key 【购买阿里服务 控制台获取填写】
          access-key: LTAI4FwRvzLckUQ2xuFE4q6N
          # 阿里secret-key 【购买阿里服务 控制台获取填写】
          secret-key: 2RmSqPLLdE1lSOqBtjIrd21kGw0O12
          # 自定义轨迹信息存储TOPIC 默认为 RMQ_SYS_TRACE_TOPIC
          customized-trace-topic: rmq_sys_TRACE_DATA_cn-qingdao-publictest
        # rocketMq 自定义消息通道配置
        bindings:
          # 阿里rocketMq binder 生产者配置
          ### 延时消息生产 producer.sync 属性需设置为true
          delay_output_channel: {producer.group: MQ_INST_XXXX_XXX%GID_QIGUANBANG_DEV, producer.sync: true}
          ### 普通生产消息
          customized_output_channel: {producer.group: MQ_INST_XXXX_XXX%GID_QIGUANBANG_DEV}
          # 阿里rocketMq binder 消费者配置
          ### 延时消息订阅
          delay_input_channel: {consumer.tags: test_delay_tag}
          ### 普通消息订阅
          customized_input_channel: {consumer.tags: test_consumer_tag}
      bindings:
        # spring cloud stream binder 生产者配置
        ### 延时消息
        delay_output_channel: {destination: MQ_INST_XXXX_XXX%common-delay-topic, content-type: application/json}
        ### 普通消息
        customized_output_channel: {destination: MQ_INST_XXXX_XXX%mg_common_topic, content-type: application/json}

        # spring cloud stream binder 消费者配置
        ### 延时消息订阅
        delay_input_channel: {destination: MQ_INST_XXXX_XXX%common-delay-topic, group: MQ_INST_XXXX_XXX%GID_AQUARIUS_DELAY, content-type: application/json}
        ### 普通消息订阅
        customized_input_channel: {destination: MQ_INST_XXXX_XXX%mg_common_topic, group: MQ_INST_XXXX_XXX%GID_QIGUANBANG_DEV, content-type: application/json}
# 服务端口号
server:
  port: 8083

# slf4j日志配置
logging:
  level:
    root: info
    com.study: debug

四.spring-alibaba-cloud-rocketmq 详细配置选项

官方文档

五.MQ消费轨迹异常

<!-- rocketMq -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.9.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-acl</artifactId>
            <version>4.9.1</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
            <!-- 排除自带rocketMq-client依赖【低版本消息无法发送成功】-->
            <exclusions>
                <exclusion>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-client</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-acl</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
举报

相关推荐

RocketMQ(十三)——实战-普通消息的发送与消费

javarabbitmq

RocketMQ系列之消息发送/消费初体验

中间件

RocketMQ-消息消费模式 顺序消费

java-rocketmqrocketmqjava

RocketMQ（七）消息消费

java-rocketmqrocketmqjava一致性hashkafka系统/运维

RocketMQ 顺序消费消息

RocketMQ面试精选

Spring Cloud Stream RabbitMQ消费消息失败

java-rabbitmqrabbitmqjavajsonspring虚拟化云计算

RocketMQ消费者没有成功消费消息的问题排查

数据重启抛出异常Linux系统/运维

RocketMQ（七）消息类型--定时/延时消息

RocketMq一些收藏Java编程语言爱好者Java 进阶Java 核心技术JVM · J...分布式系统学习

阿里云openservices rocketmq消息队列消费消息底层源码分析

java-rocketmq阿里云rocketmq构造方法.net系统/运维

RocketMQ源码解析：消息发送和消费核心流程汇总

java推送持久化负载均衡编程语言
0 条评论