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

川长思鸟来 2024-02-19 00:33 64阅读 0赞

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

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

**pom文件**

<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);
        }
        
    }

具体的结束在代码中有相应的注释，这个是最基本的实现方式，具体业务需要具体分析。