指南
中间件手册

消息队列（Redis）

blossom-spring-boot-starter-mq (opens new window)技术组件，基于 Redis 实现分布式消息队列：

使用 Stream (opens new window)特性，提供【集群】消费的能力。
使用 Pub/Sub (opens new window)特性，提供【广播】消费的能力。

疑问：什么是【广播】消费？什么是【集群】消费？

参见《阿里云 —— 集群消费和广播消费》 (opens new window)文档

# 1. 集群消费

集群消费，是指消息发送到 Redis 时，有且只会被一个消费者（应用 JVM 实例）收到，然后消费成功。如下图所示：

友情提示：

如果你需要使用到【集群】消费，必须使用 Redis 5.0.0 以上版本，因为 Stream 特性是在该版本之后才引入噢！

# 1.1 使用场景

集群消费在项目中的使用场景，主要是提供可靠的、可堆积的异步任务的能力。例如说：

短信模块，使用它异步 (opens new window)发送短信。
邮件模块，使用它异步 (opens new window)发送邮件。

相比《开发指南 —— 异步任务》 (opens new window) 来说，Spring Async 在 JVM 实例重启时，会导致未执行完的任务丢失。而集群消费，因为消息是存储在 Redis 中，所以不会存在该问题。

# 1.2 实现源码

集群消费基于 Redis Stream 实现：

实现 AbstractRedisStreamMessage 抽象类，定义【集群】消息。
使用 RedisMQTemplate 的 #send(message) 方法，发送消息。
实现 AbstractRedisStreamMessageListener 接口，消费消息。

最终使用 RedisMQAutoConfiguration 配置类，扫描所有的 AbstractRedisStreamMessageListener 监听器，初始化对应的消费者。如下图所示：

# 1.3 实战案例

以【短信发送】举例子，改造使用 Redis 作为消息队列，同时也是讲解集群消费的使用。如下图所示：

# 1.3.0 引入依赖

在 blossom-module-system-biz 模块中，引入 blossom-spring-boot-starter-mq 技术组件。如下所示：

<dependency>
    <groupId>cn.econets.boot</groupId>
    <artifactId>blossom-spring-boot-starter-mq</artifactId>
</dependency>

# 1.3.1 Message 消息

在 message 包下，修改 SmsSendMessage 类，短信发送消息。代码如下：

@Data
public class SmsSendMessage extends AbstractRedisStreamMessage { // 重点：需要继承 AbstractRedisStreamMessage 类

    /**
     * 短信日志编号
     */
    @NotNull(message = "短信日志编号不能为空")
    private Long logId;
    /**
     * 手机号
     */
    @NotNull(message = "手机号不能为空")
    private String mobile;
    /**
     * 短信渠道编号
     */
    @NotNull(message = "短信渠道编号不能为空")
    private Long channelId;
    /**
     * 短信 API 的模板编号
     */
    @NotNull(message = "短信 API 的模板编号不能为空")
    private String apiTemplateId;
    /**
     * 短信模板参数
     */
    private List<KeyValue<String, Object>> templateParams;

}

# 1.3.2 SmsProducer 生产者

在 producer 包下，修改 SmsProducer 类，Sms 短信相关消息的生产者。代码如下：

@Slf4j
@Component
public class SmsProducer {

    @Resource
    private RedisMQTemplate redisMQTemplate; // 重点：注入 RedisMQTemplate 对象

    /**
     * 发送 {@link SmsSendMessage} 消息
     *
     * @param logId 短信日志编号
     * @param mobile 手机号
     * @param channelId 渠道编号
     * @param apiTemplateId 短信模板编号
     * @param templateParams 短信模板参数
     */
    public void sendSmsSendMessage(Long logId, String mobile,
                                   Long channelId, String apiTemplateId, List<KeyValue<String, Object>> templateParams) {
        SmsSendMessage message = new SmsSendMessage().setLogId(logId).setMobile(mobile);
        message.setChannelId(channelId).setApiTemplateId(apiTemplateId).setTemplateParams(templateParams);
        redisMQTemplate.send(message); // 重点：使用 RedisMQTemplate 发送消息
    }

}

# 1.3.3 SmsSendConsumer 消费者

在 consumer 包下，修改 SmsSendConsumer 类，SmsSendMessage 的消费者。代码如下：

@Component
@Slf4j
public class SmsSendConsumer extends AbstractRedisStreamMessageListener<SmsSendMessage> { // 重点：继承 AbstractRedisStreamMessageListener 类，并填写对应的 Message 类

    @Resource
    private SmsSendService smsSendService;

    @Override // 重点：实现 onMessage 方法
    public void onMessage(SmsSendMessage message) {
        log.info("[onMessage][消息内容({})]", message);
        smsSendService.doSendSms(message);
    }

}