Springboot+Redis实现消息队列（发布订阅模式）-蒲公英云

Springboot+Redis实现消息队列（发布订阅模式）

我上找了很多例子，但是都觉得不太明白，自己琢磨了一下，做了个笔记！

因为spring对Redis进行了完整的封装，所以实现起来就比较简单。

pom文件

<!-- 整合redis -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

Redis配置文件（配置方式比较多，具体根据情况而定）：

redis:
  # Redis数据库索引（默认为0）
    database: 0
    # Redis服务器地址
    host: 127.0.0.1
    # Redis服务器连接端口
    port: 6379
    #Redis服务器连接密码（默认为空）
    password:
    #  配置多节点
    #cluster:
    #   nodes: 10.201.1.27:7001,10.201.1.27:7002,10.201.1.27:7003
    pool:
      max-active: 8 # 连接池最大连接数（使用负值表示没有限制）
      max-wait: -1  # 连接池最大阻塞等待时间（使用负值表示没有限制）
      max-idle: 8   # 连接池中的最大空闲连接
      min-idle: 0   # 连接池中的最小空闲连接
    timeout: 100    # 连接超时时间（毫秒）

1，消息发布代码

消息发布的实现是非常简单的，只需要使用 StringRedisTemplate的convertAndSend(channel, message)方法即可，

其中channel代表消息信道也可以理解为主题，message表示发布的内容。

消息发布类，模拟service层的逻辑处理

package com.boot.test1.service;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
/**
 * 消息发布类
 * @author zhh
 */
@Service
public class Test01Service {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    /**
     * 发布消息
     * @param channel 消息信道
     * @param message 消息内容
     */
    public void sendMessage(String channel, String message) {
        stringRedisTemplate.convertAndSend(channel, message);
    }
    /**
     * 发布消息的方法
     */
    public void setStr01(){
        this.sendMessage("mq_01", "发送信息内容01");
        this.sendMessage("mq_01", "发送信息内容011");
        this.sendMessage("mq_02", "发送信息内容02");
    }
}

2，消息接收、处理代码

2.1 消息监听注册配置

主要把消息监听注册到容器里面

package com.boot.common.conf;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import com.boot.test1.redismq.MessageReceiver;
/**
 * Redis消息处理配置类
 * @author zhh
 * @date 2018-09-30
 *
 */
@Configuration
public class RedisMQConfig {
    /**
     * 注入消息监听容器
     * @param connectionFactory 连接工厂
     * @param listenerAdapter   监听处理器1
     * @param listenerAdapter   监听处理器2 (参数名称需和监听处理器的方法名称一致，因为@Bean注解默认注入的id就是方法名称)
     * @return
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
            MessageListenerAdapter listenerAdapter,
            MessageListenerAdapter listenerAdapter2) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //订阅一个叫mq_01 的信道
        container.addMessageListener(listenerAdapter, new PatternTopic("mq_01"));
        //订阅一个叫mq_02 的信道
        container.addMessageListener(listenerAdapter2, new PatternTopic("mq_02"));
        //这个container 可以添加多个 messageListener
        return container;
    }
    /**
     * 消息监听处理器1
     * @param receiver 处理器类
     * @return
     */
    @Bean
    MessageListenerAdapter listenerAdapter(MessageReceiver receiver) {
        //给messageListenerAdapter 传入一个消息接收的处理器，利用反射的方法调用“receiveMessage”
        return new MessageListenerAdapter(receiver, "receiveMessage"); //receiveMessage：接收消息的方法名称
    }
    /**
     * 消息监听处理器2
     * @param receiver 处理器类
     * @return
     */
    @Bean
    MessageListenerAdapter listenerAdapter2(MessageReceiver receiver) {
        //给messageListenerAdapter 传入一个消息接收的处理器，利用反射的方法调用“receiveMessage2”
        return new MessageListenerAdapter(receiver, "receiveMessage2"); //receiveMessage：接收消息的方法名称
    }
}

2.2 自定义的消息处理器

package com.boot.test1.redismq;
import org.springframework.stereotype.Component;
/**
 * MQ消息处理器
 * @author zhh
 */
@Component
public class MessageReceiver {
    /**
     * 接收消息的方法1
     **/
    public void receiveMessage(String message){
        System.out.println("receiveMessage接收到的消息："+message);
    }
    /**
     * 接收消息的方法2
     **/
    public void receiveMessage2(String message){
        System.out.println("receiveMessage2接收到的消息："+message);
    }
}