消息队列组件 (Messaging)

API文档

https://tech.rydeen.com.cn/apidocs/messaging/4.6.0/index.html

组件作用

项目开发中，消息队列中间件是一个高频使用的技术中间件，而不同的客户会有不同的消息队列中间件的选择，而不同的中间件会有不同的连接驱动代码。通常情况下，我们要么通过设计模式在代码中自行编码屏蔽中间件差异，要么将业务代码拆分多个版本，每个版本对应一个中间件的连接驱动代码。这两种方式都会增加代码的复杂度，降低代码的可读性和可维护性。

为了解决该问题，我们设计了消息队列组件，该组件提供了统一的消息队列接口，屏蔽了不同消息队列中间件的差异，使得业务代码可以不用关心消息队列中间件的选择，将所有差异屏蔽到配置文件中，代码中提供统一的消息生产者和消费者的接口进行调用，通过调整配置文件的内容来达到对多消息队列中间件的支持，从而提高了代码的可读性和可维护性。

目前我们支持的消息队列中间件有：RabbitMQ、Kafka、Kafka Stream、RocketMQ、EventHubs、GCP-PubSub、Kinesis、ServiceBus、SNS、SQS。基本已经覆盖了常见和不常见的消息队列中间件。足以覆盖我们99%的客户，如果有不支持的消息队列中间件，我们可以通过自定义中间件扩展的方式来支持。当需要更换中间件支持的时候，我们仅仅需要引入rydeen-component-messaging核心组件包以及对应的中间件支持包即可，消息队列组件包含的包清单如下：

核心包：

• rydeen-component-messaging

支持包：

• rydeen-component-messaging-eventhubs
• rydeen-component-messaging-gcp-pubsub
• rydeen-component-messaging-kafka
• rydeen-component-messaging-kinesis
• rydeen-component-messaging-rabbitmq
• rydeen-component-messaging-rocketmq
• rydeen-component-messaging-servicebus
• rydeen-component-messaging-sns
• rydeen-component-messaging-solace
• rydeen-component-messaging-sqs

组件用法

引入依赖

消息队列组件配置方式如下：

首先在项目所属的Maven模块中引入组件坐标：

:::success 版本提示

该组件的包依赖分拆自4.2.0版本开始可用 :::

为了减小打包后的jar包体积，从4.2.0版本开始，我们对当前组件按照MQ中间件类型对依赖进行了分拆，从4.1.0的

<dependency>
    <groupId>cn.com.rydeen.component</groupId>
    <artifactId>rydeen-component-messaging</artifactId>
</dependency>

分拆成

<dependencies>
    <!-- 主服务包（必须，其余的按需引入） -->
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-eventhubs</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-gcp-pubsub</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-kinesis</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-rabbitmq</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-rocketmq</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-servicebus</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-sns</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-solace</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.com.rydeen.component</groupId>
        <artifactId>rydeen-component-messaging-sqs</artifactId>
    </dependency>
</dependencies>

配置文件

编写配置文件，消息队列组件基于spring-cloud-stream，配置文件的结构主要分为两部分，一部分是针对消息队列中间件的binder配置，用来指定消息队列中间件的服务器配置信息，以及中间件内Topic、Queue或Exchange的配置信息，另一部分是针对消息队列的binding配置，用来指定消息队列的生产者和消费者的配置信息。

想要理解消息队列组件的配置文件，首先需要了解spring-cloud-stream组件的几个基本概念：

• binder：消息队列中间件的服务器配置信息，以及中间件内Topic、Queue或Exchange的配置信息。
• binding：消息队列的生产者和消费者的配置信息。
• Input：输入管道，用于接收消息（即：消息消费者）。
• Output：输出管道，用于发送消息（即：消息生产者）。

本组件支持绑定1个或多个消息队列中间件，每个中间件的配置信息都需要在binder节点下进行配置，每个中间件的binder配置都会映射一个binderName，而每个消息队列的生产者和消费者的配置信息都需要在binding节点下进行配置，每个binding都需要和一个binder进行绑定（当仅有1个binder时例外），以确定当前binding是哪一个binder的配置，下面是一个消息队列组件的配置文件示例：

:::note binding 服务属性配置

查看官方文档 :::

:::note binding 属性配置

查看官方文档 :::

:::note binder 配置

binder配置和具体中间件包有关，具体请参考官方文档或扩展支持的对应章节：查看官方文档 :::

:::note producer 配置

producer 消息生产者配置请参考官方文档：查看官方文档 :::

:::note consumer 配置

consumer 消息消费者配置请参考官方文档：查看官方文档 :::

注意：这里我们不会去解释官方文档中已经明确告知的基本配置信息，请自行查看官方文档或查看本文档中在配置文件示例中已经解释的内容，其余内容请自行学习。

RabbitMQ配置示例（仅用于演示）

spring:
    main:
        allow-bean-definition-overriding: true

    # Cloud 配置
    cloud:
        function:
            definition: normalProcess;delayProcess
        stream:
            # 默认binder器
            default-binder: rabbit
            # 幂等去重服务使用的数据源类型(messaging组件自定义扩展功能)
            datasource: memory
            # RabbitMQ配置
            rabbit:
                bindings:
                    demo1-out-0:
                        producer:
                            delayed-exchange: false
                            exchange-type: topic
                            queue-name-group-only: true
                            exchange-durable: true
                    demo2-out-0:
                        producer:
                            delayed-exchange: true
                            exchange-type: topic
                            queue-name-group-only: true
                    delayProcess-in-0:
                        consumer:
                            delayed-exchange: true
                            exchange-type: topic
                            queue-name-group-only: true
                    normalProcess-in-0:
                        consumer:
                            delayed-exchange: false
                            exchange-type: topic
                            queue-name-group-only: true
            binders:
                # 实例别名，在bindings里面的in、out中进行引用，指定对应的topic使用的实例
                rabbit:
                    type: rabbit
                    inherit-environment: true
                    default-candidate: true
                    environment:
                        spring:
                            rabbitmq:
                                host: 10.100.0.87
                                port: 5672
                                username: admin
                                password: admin
                                virtual-host: /
            # binding 相关配置
            bindings:
                # OUTPUT：交换器1输出通道
                demo1-out-0:
                    binder: rabbit
                    # 指定当前队列所属Exchange
                    destination: demo1Exchange
                    # 指定当前队列的消息类型
                    content-type: application/json
                    # 指定当前队列的消息组(即RabbitMQ的Queue名称)
                    group: normalGroup
                # OUTPUT：交换器2输出通道
                demo2-out-0:
                    binder: rabbit
                    # 指定当前队列所属Exchange
                    destination: demo2Exchange
                    # 指定当前队列的消息类型
                    content-type: application/json
                    # 指定当前队列的消息组(即RabbitMQ的Queue名称)
                    group: delayGroup
                # INPUT：默认输入通道定义（如需多通道请复制该节点后修改自定义节点）
                # 此处不要修改该INPUT，如果需要在项目中使用多个消费者监听，请
                # 在下方自行添加消费者监听节点，此节点为默认消费者监听器，平台
                # 组件提供了一套默认API
                normalProcess-in-0:
                    binder: rabbit
                    # 指定当前队列所属Exchange
                    destination: demo1Exchange
                    # 指定当前队列的消息类型
                    content-type: application/json
                    # 当前消费者作为公共组的消费者，其他业务组请勿使用该Group
                    group: normalGroup
                    # 消费者相关配置
                    consumer:
                        # 最大尝试次数
                        max-attempts: 3
                        # 重试起始间隔时间
                        back-off-initial-interval: 500
                        # 重试最大间隔时间
                        back-off-max-interval: 1000
                        # 重试间隔时间乘数
                        back-off-multiplier: 2.0
                        # 消费者并发数
                        concurrency: 10
                delayProcess-in-0:
                    binder: rabbit
                    # 指定当前队列所属Exchange
                    destination: demo2Exchange
                    # 指定当前队列的消息类型
                    content-type: application/json
                    # 当前消费者作为公共组的消费者，其他业务组请勿使用该Group
                    group: delayGroup
                    # 消费者相关配置
                    consumer:
                        # 最大尝试次数
                        max-attempts: 3
                        # 重试起始间隔时间
                        back-off-initial-interval: 500
                        # 重试最大间隔时间
                        back-off-max-interval: 1000
                        # 重试间隔时间乘数
                        back-off-multiplier: 2.0
                        # 消费者并发数
                        concurrency: 10
management:
    health:
        rabbit:
            enabled: false

Kafka配置示例（仅用于演示）

spring:
    main:
        allow-bean-definition-overriding: true
    # Cloud 配置
    cloud:
        function:
            definition: normalProcess
        # Kafka 配置
        stream:
            # 默认binder器（此处支持kafka、RabbitMQ、RocketMQ，但是配置文件内容有少许差异，请参考各队列各自的配置文件）
            default-binder: kafka
            # 幂等去重服务使用的数据源类型
            datasource: memory
            # Kafka配置
            kafka:
                binder:
                    # kafka服务器地址
                    brokers: 192.168.4.199:9092
                    # 是否自动创建主题
                    auto-create-topics: true
            # binding 相关配置
            bindings:
                # OUTPUT：主题1输出通道
                demo1-out-0:
                    destination: demo1-topic
                    content-type: application/json
                    # 生产者参数配置
                    producer:
                        # 指定该生产者是否自动启动（默认：true）
                        auto-startup: true
                        # 生产者配置请参考 org.springframework.cloud.stream.binder.ProducerProperties

                # OUTPUT：主题2输出通道
                demo2-out-0:
                    destination: demo2-topic
                    content-type: application/json
                    # 生产者参数配置
                    producer:
                        # 指定该生产者是否自动启动（默认：true）
                        auto-startup: true
                        # 生产者配置请参考 org.springframework.cloud.stream.binder.ProducerProperties

                # INPUT：默认输入通道定义（如需多通道请复制该节点后修改自定义节点）
                # 此处不要修改该INPUT，如果需要在项目中使用多个消费者监听，请
                # 在下方自行添加消费者监听节点，此节点为默认消费者监听器，平台
                # 组件提供了一套默认API
                normalProcess-in-0:
                    destination: demo1-topic,demo2-topic
                    content-type: application/json
                    # 消费者相关配置
                    consumer:
                        # 最大尝试次数
                        max-attempts: 3
                        # 重试起始间隔时间
                        back-off-initial-interval: 500
                        # 重试最大间隔时间
                        back-off-max-interval: 1000
                        # 重试间隔时间乘数
                        back-off-multiplier: 2.0
                        # 更多配置参考 org.springframework.cloud.stream.binder.ConsumerProperties

RocketMQ配置示例（仅用于演示）

spring:
    main:
        allow-bean-definition-overriding: true
    # Cloud 配置
    cloud:
        function:
            definition: normalProcess
        # Kafka 配置
        stream:
            # 默认binder器（此处支持kafka、RabbitMQ、RocketMQ，但是配置文件内容有少许差异，请参考各队列各自的配置文件）
            default-binder: rocketmq
            # 幂等去重服务使用的数据源类型
            datasource: memory
            # RocketMQ 配置
            rocketmq:
                binder:
                    # 是否启用队列
                    enabled: true
                    # 命名服务器地址
                    name-server: 10.100.0.107:9876
#                    # 用户名（当使用阿里云PaaS服务时配置）
#                    access-key:
#                    # 密码（当使用阿里云PaaS服务时配置）
#                    secret-key:
                    # 对消费者或生产者进行分组的组名
                    group: menu_queue
                    # 用于 RocketMQ 在数据存储层面，进行多租户数据存储的命名空间，默认为空，可选值
                    namespace:
                    # 用于构建 ClientID 的单位名称，默认为空，可选值
                    unit-name:
                    # 服务访问类型
                    # local：IDC机房的自建服务
                    # cloud：云 PaaS 服务（目前针对阿里云的 RocketMQ、华为云和腾讯云待确认）
                    access-channel: cloud
                    # 获取 Topic 路由信息的间隔时间（默认：30000ms）
                    poll-name-server-interval: 30000
                    # 客户端和 Broker 之间的心跳间隔时间（默认：30000ms）
                    heartbeat-broker-interval: 30000
                    # 持久化消费者间隔时间（默认：5000ms）
                    persist-consumer-offset-interval: 5000
                    # 是否使用 VIP 通道（VIP 通道和非 VIP 通道的区别是使用不同的端口号进行通信）
                    vip-channel-enabled: false
                    # 是否启用消息追踪（默认：true）
                    enable-msg-trace: true
                    # 自定义追踪的主题（默认 null）
                    customized-trace-topic: custom-topic
                    # 是否启用 TLS 协议（默认：false）
                    use-t-l-s: false

            # binding 相关配置
            bindings:
                # OUTPUT：主题1输出通道
                demo1-out-0:
                    destination: demo1-topic
                    content-type: application/json
                    # 生产者参数配置
                    producer:
                        # 指定该生产者是否自动启动（默认：true）
                        auto-startup: true


                        # 生产者配置请参考 org.springframework.cloud.stream.binder.ProducerProperties

                # OUTPUT：主题2输出通道
                demo2-out-0:
                    destination: demo2-topic
                    content-type: application/json
                    # 生产者参数配置
                    producer:
                        # 指定该生产者是否自动启动（默认：true）
                        auto-startup: true
                        # 生产者配置请参考 org.springframework.cloud.stream.binder.ProducerProperties

                # INPUT：默认输入通道定义（如需多通道请复制该节点后修改自定义节点）
                # 此处不要修改该INPUT，如果需要在项目中使用多个消费者监听，请
                # 在下方自行添加消费者监听节点，此节点为默认消费者监听器，平台
                # 组件提供了一套默认API
                normalProcess-in-0:
                    destination: demo1-topic,demo2-topic
                    content-type: application/json
                    # 消费者相关配置
                    consumer:
                        # 最大尝试次数
                        max-attempts: 3
                        # 重试起始间隔时间
                        back-off-initial-interval: 500
                        # 重试最大间隔时间
                        back-off-max-interval: 1000
                        # 重试间隔时间乘数
                        back-off-multiplier: 2.0
                        # 更多配置参考 org.springframework.cloud.stream.binder.ConsumerProperties
management:
    health:
        rabbit:
            enabled: false

多MQ中间件的配置示例（仅用于演示）

官方文档说明，请参考：官方文档

# 重要说明：混合多数据源配置方案，当项目中存在多个队列时，请使用该配置方案
# 该配置方案支持混合队列，包括RabbitMQ、Kafka、RocketMQ进行组合使用，请
# 按照下方配置方式拷贝到项目以后，适当的进行修改就可以使用。
spring:
  main:
    allow-bean-definition-overriding: true
  # Cloud 配置
  cloud:
    function:
      definition: normalProcess
    # Kafka 配置
    stream:
      # 幂等去重服务使用的数据源类型
      datasource: memory
      # 消息处理失败后的最大重试次数（默认：3次）
      max-retries: 3
      # binding 相关配置
      bindings:
        # OUTPUT：餐厅主档数据更新通知
        demo1-out-0:
          destination: queue-demo1
          content-type: application/json
          binder: rabbit
        # OUTPUT：餐厅主档数据更新通知
        demo2-out-0:
          destination: queue-demo2
          content-type: application/json
          binder: kafka
        # OUTPUT：组织数据更新通知
        # INPUT：任务模块输入通道定义
        normalProcess-in-0:
          destination: queue-demo1,queue-demo2
          content-type: application/json
          # 消费者相关配置
          consumer:
            # 最大尝试次数
            max-attempts: 3
            # 重试起始间隔时间
            back-off-initial-interval: 500
            # 重试最大间隔时间
            back-off-max-interval: 1000
            # 重试间隔时间乘数
            back-off-multiplier: 2.0
      # 多队列方案
      binders:
        # MQ1名称：名称可自定义
        rabbit:
          type: rabbit
          inherit-environment: true
          default-candidate: true
          environment:
            spring:
              rabbitmq:
                host: 10.20.1.133
                port: 5672
                username: richie696
                password: 111111
        # MQ2名称：名称可自定义
        kafka:
          # 当前绑定器对应的MQ队列类型
          type: kafka
          # 当前绑定器是否继承应用程序自身的环境配置
          inherit-environment: true
          # 当前绑定器配置是默认绑定器的候选配置，还是只能在显式引用时使用（默认：true）
          # 当需要让当前配置不影响默认配置时，可以将该属性设置为 false
          default-candidate: true
          # MQ队列的环境参数配置
          environment:
            spring:
              kafka:
                # broker地址
                bootstrap-servers: 127.0.0.1
                # 生产者配置
                producer:
                  # 重试次数
                  retries: 3
                  # 应答级别：多少个分区副本备份完成时向生产者发送ACK确认
                  # ACKS参数含义：
                  # acks = 0：
                  # 生产者不等待broker对消息的确认，只要将消息放到缓冲区，就认为消息已经发送完成。
                  # 该情形不能保证broker是否真的收到了消息，retries配置也不会生效。发送的消息的返
                  # 回的消息偏移量永远是-1。
                  # acks = 1
                  # 表示消息只需要写到主分区即可，然后就响应客户端，而不等待副本分区的确认。在该
                  # 情形下，如果主分区收到消息确认之后就宕机了，而副本分区还没来得及同步该消息，
                  # 则该消息丢失。
                  # acks = -1
                  # 首领分区会等待所有的ISR副本分区确认记录。该处理保证了只要有一个ISR副本分区存
                  # 活，消息就不会丢失。这是Kafka最强的可靠性保证。
                  acks: 1
                  # 批量大小
                  batch-size: 16384
                  properties:
                    linger:
                      # 提交延时：当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
                      # linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了
                      ms: 0
                    # 自定义分区器
                    partitioner:
                      # 该分区器为Kafka默认的分区器，如果无需自定义请删除该配置节点
                      class: org.apache.kafka.clients.producer.internals.DefaultPartitioner
                  # 生产端缓冲区大小
                  buffer-memory: 33554432
                  # KEY序列化器（可自定义，如果使用默认可删除该配置）
                  key-serializer: org.apache.kafka.common.serialization.StringSerializer
                  # VALUE 序列化器（可自定义，如果使用默认可删除该配置）
                  value-serializer: org.apache.kafka.common.serialization.StringSerializer
                # 消费者配置
                consumer:
                  properties:
                    group:
                      # 默认的消费组ID
                      id: default-group
                    # 消费请求超时时间（单位：毫秒）
                    request:
                      timeout:
                        ms: 180000
                    # 消费会话超时时间（单位：毫秒，超过这个时间 consumer 没有发送心跳，就会触发
                    # Rebalance 操作，该操作会停止一切队列消费操作，等待再平衡结束以后才会恢复消
                    # 息的发送。）
                    session:
                      timeout:
                        ms: 120000
                  # 是否自动提交消费者偏移量（默认：true）
                  enable-auto-commit: true
                  # 消费者偏移量自动提交到 Kafka 的频率（单位：毫秒，接收到消息后多久提交 offset）
                  auto-commit-interval: 1000
                  # 当 Kafka 中没有初始偏移量或者当前偏移量超出服务器上的偏移量上限时的处理方式
                  # earliest：重置为分区中最小的 offset
                  # latest：重置为分区中最新的 offset
                  # none：只要有一个分区不存在已提交的 offset 就抛出异常
                  # exception：直接抛出异常
                  auto-offset-reset: latest
                  # KEY反序列化器（可自定义，如果使用默认可删除该配置）
                  key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
                  # VALUE 反序列化器（可自定义，如果使用默认可删除该配置）
                  value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
                  # 批量消费时，每次最多消费多少条消息
                  max-poll-records: 50
                  # 消费者心跳包发送到消费者协调器的心跳间隔（单位：毫秒）
                  # 如果无特殊需求，可以不指定
                  heartbeat-interval: 30000
                listener:
                  # 消费端监听的 topic 不存在时，项目启动直接报错，通常建议关闭。
                  missing-topics-fatal: false
                  # 设置消费者消费方式
                  # batch：每次消费1个批次
                  # single：每次消费1条
                  type: batch
                # 更多配置请查看：org.springframework.boot.autoconfigure.kafka.KafkaProperties

实际生产使用，请根据实际情况做调整。

消息生产者

在完成上一个章节的配置后，我们就可以在项目中引入消费生产者（即：发送者）接口进行消息发送了，我们为生产者提供了MessageService接口，用于简化发送流程，该接口一共有2个实现，一个实现发送的是一个消息事件对象，而另一个发送的是消息的原始报文，用于不同场景，可自行选择使用。这两个实现的类型别名分别是：defaultMessageService(消息事件)和originMessageService(原始消息)。

下面我们就介绍一下cn.com.rydeen.component.messaging.service.MessageService接口的功能。

package cn.com.rydeen.component.messaging.service;

import org.springframework.util.MimeType;

/**
 * MQ消息队列服务接口
 *
 * @author richie696
 * @version 1.0
 * @since 2022-09-27 11:15:18
 */
public interface MessageService {

    /**
     * 采用通道名称作为默认消息键的队列发送方法
     *
     * @param topicAlias 发送消息的Topic别名（建议使用枚举进行定义，不要硬编码字符串）
     * @param message    发送的具体消息
     * @return 返回发送结果（true：发送成功，false：发送失败）
     */
    boolean sendMessage(String topicAlias, Object message);

    /**
     * 采用通道名称作为默认消息键的队列发送方法（多MQ服务时使用）
     *
     * @param topicAlias 发送消息的Topic别名（建议使用枚举进行定义，不要硬编码字符串）
     * @param binderName 绑定器名称（多MQ服务时使用）
     * @param message    发送的具体消息
     * @return 返回发送结果（true：发送成功，false：发送失败）
     */
    boolean sendMessage(String topicAlias, String binderName, Object message);

    /**
     * 采用通道名称作为默认消息键的队列发送方法（可自定义OUTPUT消息的类型）
     *
     * @param topicAlias        发送消息的Topic别名（建议使用枚举进行定义，不要硬编码字符串）
     * @param message           发送的具体消息
     * @param outputContentType 输出的消息体类型
     * @see org.springframework.util.MimeTypeUtils 输出类型可使用“MimeTypeUtils”中的常量类型
     * @return 返回发送结果（true：发送成功，false：发送失败）
     */
    boolean sendMessage(String topicAlias, Object message, MimeType outputContentType);

    /**
     * 可自定义消息键的消息队列发送方法（多MQ服务时使用）
     *
     * @param topicAlias        发送消息的Topic别名（建议使用枚举进行定义，不要硬编码字符串）
     * @param binderName    绑定器名称（多MQ服务时使用）
     * @param message           发送的具体消息
     * @param outputContentType 输出的消息体类型
     * @see org.springframework.util.MimeTypeUtils 输出类型可使用“MimeTypeUtils”中的常量类型
     * @return 返回发送结果（true：发送成功，false：发送失败）
     */
    boolean sendMessage(String topicAlias, String binderName, Object message, MimeType outputContentType);

    /**
     * 采用通道名称作为默认消息键的队列发送方法
     *
     * @param topicAlias 发送消息的Topic别名（建议使用枚举进行定义，不要硬编码字符串）
     * @param message    发送的具体消息
     * @param delayTime 延迟时间（单位：毫秒）
     * @return 返回发送结果（true：发送成功，false：发送失败）
     */
    boolean sendMessage(String topicAlias, Object message, long delayTime);

    /**
     * 采用通道名称作为默认消息键的队列发送方法（多MQ服务时使用）
     *
     * @param topicAlias 发送消息的Topic别名（建议使用枚举进行定义，不要硬编码字符串）
     * @param binderName 绑定器名称（多MQ服务时使用）
     * @param message    发送的具体消息
     * @param delayTime 延迟时间（单位：毫秒）
     * @return 返回发送结果（true：发送成功，false：发送失败）
     */
    boolean sendMessage(String topicAlias, String binderName, Object message, long delayTime);

    /**
     * 采用通道名称作为默认消息键的队列发送方法（可自定义OUTPUT消息的类型）
     *
     * @param topicAlias        发送消息的Topic别名（建议使用枚举进行定义，不要硬编码字符串）
     * @param message           发送的具体消息
     * @param outputContentType 输出的消息体类型
     * @param delayTime 延迟时间（单位：毫秒）
     * @see org.springframework.util.MimeTypeUtils 输出类型可使用“MimeTypeUtils”中的常量类型
     * @return 返回发送结果（true：发送成功，false：发送失败）
     */
    boolean sendMessage(String topicAlias, Object message, MimeType outputContentType, long delayTime);

    /**
     * 可自定义消息键的消息队列发送方法（多MQ服务时使用）
     *
     * @param topicAlias        发送消息的Topic别名（建议使用枚举进行定义，不要硬编码字符串）
     * @param binderName    绑定器名称（多MQ服务时使用）
     * @param body           发送的具体消息
     * @param outputContentType 输出的消息体类型
     * @param delayTime 延迟时间（单位：毫秒）
     * @see org.springframework.util.MimeTypeUtils 输出类型可使用“MimeTypeUtils”中的常量类型
     * @return 返回发送结果（true：发送成功，false：发送失败）
     */
    boolean sendMessage(String topicAlias, String binderName, Object body, MimeType outputContentType, long delayTime);
}

根据接口定义中的说明，我们可以发现消息发送接口同时支持实时消息和延时消息的发送（目前需要MQ中间件的支持，暂未实现本地计划任务的延迟队列），当遇到不支持延迟队列的MQ时，延迟队列接口会以实时消息的方式进行发送。

消息消费者

和消息生产者一样，消息消费者也提供了默认的实现，但是使用上和生产者有所区别，我们从最简单的开始，当我们的应用中仅有一个消费者通道时，我们可以用messaging组件提供的cn.com.rydeen.component.messaging.consumer.BaseConsumer进行消费。

消息消费者中我们全局实现了幂等去重、消息消费时间统计、消息体报文自动转换、消息重新压入队列等诸多功能，而这些功能对消费者端都是透明的，无需关心。而只有幂等去重的功能是支持配置的，可以选择在单机下通过内存的方式幂等，或者在多实例部署的集群环境下通过Redis幂等，以达到不同场景幂等的效果。对幂等的配置参数如下：

spring:
    cloud:
        stream:
            # 幂等去重服务使用的数据源类型（支持：memory和redis）
            datasource: memory

单管道消费者的使用

所谓全局消费者，指的是在当前应用中，我们仅需要唯一的一个消费者管道进行不同类型的消息消费而无需注册多个管道，进行消息的管道隔离时，我们就可以使用这种方式进行消费者的注册，这种方式是最简单的，也是最适用于大多数消费场景的使用方式。具体使用方式如下：

package cn.com.rydeen.component.template.messaging.consumer;

import cn.com.rydeen.base.utils.data.JsonUtils;
import cn.com.rydeen.component.template.messaging.domain.UserInfo;
import cn.com.rydeen.component.template.messaging.service.UserService;
import cn.com.rydeen.component.messaging.consumer.BaseConsumer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;

import static cn.com.rydeen.component.template.messaging.constants.TopicConstant.*;

/**
 * 用户消费者类
 *
 * @author richie696
 * @version 1.0
 * @since 2022-10-02 11:32:52
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class UserConsumer {

    private final BaseConsumer consumer;

    private final UserService userService;

    @PostConstruct
    public void initial() {
        // 普通消息队列
        consumer.registerConsumer("demo1-out-0", message -> {
            log.info("收到消息：{}", JsonUtils.getInstance().serialize(message));
            String value = userService.sayHello(new String(message.getContent()));
            log.info("普通消息：{}", value);
            return true;
        });

        // 延时消息队列
        consumer.registerConsumer("demo2-out-0", message -> {
            log.info("收到消息：{}", JsonUtils.getInstance().serialize(message));
            String msg = message.getBody(String.class);
            log.info("延迟消息：{}，延迟时间：{} 毫秒", msg, message.getDelayTime());
            return true;
        });

        // 延时消息队列
        consumer.registerConsumer("demo3-out-0", message -> {
            log.info("收到消息：{}", JsonUtils.getInstance().serialize(message));
            UserInfo userInfo = message.getBody(UserInfo.class);
            String value = userService.sayHello(userInfo);
            log.info("系统和你打了招呼：{}", value);
            return true;
        });
    }
}

至此，我们就完成了消费者队列的订阅。

相信大家已经注意到，消费者注册代码中的第一个入参的名称和配置文件中binding的输出管道别名是一致的。是的，我们就是通过生产者的binding别名与消费者的消费实例进行绑定的。当绑定成功后，在当前实例收到消息后，会经过一系列预处理和加工，将收到的消息通知到此处注册的消费者对象**（注意：此处的消费者实例仅适用于defaultMessageService服务实现，原始消息服务请参考BaseConsumer内部实现自行编写）**），这样就完成了一个完整的消费过程。

多管道消费者的使用

在某些情况下，一个应用实例中可能会有多个消费管道的问题，比如：系统中有多个MQ、多个Exchange、Queue、Topic等特殊情况，无法公用时，我们就需要对这些情况进行消费者隔离。此时，我们就无法直接使用BaseConsumer来接收消息了，我们需要扩展messaging组件的消费者处理器，方法如下：

/* 用于接收订单数据的消费者处理器 */

package cn.com.rydeen.component.messaging.consumer;

import cn.com.rydeen.component.messaging.config.MessagingProperties;
import cn.com.rydeen.component.messaging.event.MessageEvent;
import cn.com.rydeen.component.messaging.filter.handler.MessageHandlerService;
import cn.com.rydeen.component.messaging.service.impl.MessageServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.function.Function;

/**
 * 订单接收消息队列消费者处理器
 *
 * @author richie696
 * @version 1.0
 * @since 2024/03/11
 */
@Slf4j
@Component
public class OrderConsumer extends AbstractBaseConsumer {

    public OrderConsumer(MessageHandlerService messageHandlerService,
                        MessageServiceImpl messageService,
                        @Qualifier("messagingProperties")MessagingProperties properties) {
        super(messageHandlerService, messageService, properties);
    }

    @Bean
    public Function<Message<MessageEvent>, MessageEvent> orderProcess() {
        return getMessageFunction(false);
    }

}

/* 用于从外部系统做数据同步的消费者处理器 */

package cn.com.rydeen.component.messaging.consumer;

import cn.com.rydeen.component.messaging.config.MessagingProperties;
import cn.com.rydeen.component.messaging.event.MessageEvent;
import cn.com.rydeen.component.messaging.filter.handler.MessageHandlerService;
import cn.com.rydeen.component.messaging.service.impl.MessageServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.function.Function;

/**
 * 外部系统数据同步队列消费者处理器
 *
 * @author richie696
 * @version 1.0
 * @since 2021/08/11
 */
@Slf4j
@Component
public class DataSyncConsumer extends AbstractBaseConsumer {

    public DataSyncConsumer(MessageHandlerService messageHandlerService,
                        MessageServiceImpl messageService,
                        @Qualifier("messagingProperties")MessagingProperties properties) {
        super(messageHandlerService, messageService, properties);
    }

    @Bean
    public Function<Message<MessageEvent>, MessageEvent> dataSyncProcess() {
        return getMessageFunction(true);
    }
}

当完成上述消费者处理器的创建后，需要将这些处理器注册到application.yml配置文件中，如下：

spring:
    cloud:
        function:
            definition: orderProcess, dataSyncProcess
        stream:
            bindings:
              
                # 其它配置省略...
              
                dataSyncProcess-in-0:
                    destination: data-sync-topic
                    content-type: application/json
                    consumer:
                        max-attempts: 3
                        back-off-initial-interval: 500
                        back-off-max-interval: 1000
                        back-off-multiplier: 2.0
                        concurrency: 10
                        # 更多配置参考 org.springframework.cloud.stream.binder.ConsumerProperties
                        
                orderProcess-in-0:
                    destination: order-receive-topic
                    content-type: application/json
                    consumer:
                        max-attempts: 3
                        back-off-initial-interval: 500
                        back-off-max-interval: 1000
                        back-off-multiplier: 2.0
                        concurrency: 10
                        # 更多配置参考 org.springframework.cloud.stream.binder.ConsumerProperties

然后，对应的消费队列监听器对象分别和这2个消费者处理器进行绑定和注册，即可完成多管道消费者的配置和使用过程。