SpringBoot整合Redis消息队列

女爷i 2023-02-21 01:58 36阅读 0赞

# 消息模型 #

## 1）队列模型 ##

队列模型如图所示，它具有以下几个特点，就像我们用微信和好友（群聊除外）聊天一样，微信就是这个队列，我们可以和很多个好友聊天，但是每条消息只能发给一个好友。

*  只有一个消费者将获得消息
 *  生产者不需要在接收者消费该消息期间处于运行状态，接收者也同样不需要在消息发送时处于运行状态。
 *  每一个成功处理的消息都由接收者签收

![队列模型][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h4YzE2MDU2Mjk4OTU_size_16_color_FFFFFF_t_70]

## 2）发布/订阅模型 ##

发布/订阅模型如图所示，不用说，和订阅公众号是一样的。

*  多个消费者可以获得消息
 *  在发布者和订阅者之间存在时间依赖性。发布者需要建立一个topic，以便客户能够购订阅。订阅者必须保持持续的活动状态以接收消息，除非订阅者建立了持久的订阅。在那种情况下，在订阅者未连接时发布的消息将在订阅者重新连接时重新发布

![发布订阅模型][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h4YzE2MDU2Mjk4OTU_size_16_color_FFFFFF_t_70 1]

# Redis实现 #

1.  对于队列模型，我们可以使用redis的list数据结构，通过LPUSH和RPOP来实现一个队列。
2.  发布/订阅模型就更简单了，redis官方就支持，而且还可以使用PSUBSCRIBE支持模式匹配，使用如下命令，即可订阅所有f开头的订阅，具体可查看文档。

PSUBSCRIBE f*

3. keyspace notifications（键空间通知）

该功能是在redis2.8之后引入的，即客户端可以通过pub/sub机制，接收key的变更的消息。换句话说，就是redis官方提供了一些topic，帮助我们去监听redis数据库中的key，我曾经就使用其中的‘keyevent@0:expired’实现了定时任务。

# 和SpringBoot整合 #

首先得介绍一下spring-data-redis中的两种template的默认Serializer，当然Spring还提供其他的序列化器，具体可查看文档，也可以自己实现RedisSerializer接口，构建自己的序列化器。

<table> 
 <thead> 
  <tr> 
   <th>template</th> 
   <th>default serializer</th> 
   <th>serialization</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>RedisTemplate</td> 
   <td>JdkSerializationRedisSerializer</td> 
   <td>序列化String类型的key和value</td> 
  </tr> 
  <tr> 
   <td>StringRedisTemplate</td> 
   <td>StringRedisSerializer</td> 
   <td>使用Java序列化</td> 
  </tr> 
 </tbody> 
</table>

## 消息队列模型 ##

消息队列，这个就需要自己造轮子了，在Spring中使用redisTemlate操作数据库，而对于不同的数据类型则需要不同的操作方式，如下表格所示，具体还是请看[官方文档][Link 1]。

实现队列选择list数据结构，redisTemplate.opsForList()使用起来非常简单，和redis命令基本一致。

<table> 
 <thead> 
  <tr> 
   <th>数据类型</th> 
   <th>操作方式</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>string</td> 
   <td>redisTemplate.opsForValue()</td> 
  </tr> 
  <tr> 
   <td>hash</td> 
   <td>redisTemplate.opsForHash()</td> 
  </tr> 
  <tr> 
   <td>list</td> 
   <td>redisTemplate.opsForList()</td> 
  </tr> 
  <tr> 
   <td>set</td> 
   <td>redisTemplate.opsForSet()</td> 
  </tr> 
 </tbody> 
</table>

1.先定义一个消息的POJO（MessageEntity实体类）

2.配置spring data redis

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
    import org.springframework.data.redis.serializer.StringRedisSerializer;
    
    import com.fasterxml.jackson.annotation.JsonAutoDetect;
    import com.fasterxml.jackson.annotation.PropertyAccessor;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.johnfnash.learn.redis.queue.entity.MessageEntity;
    
    @Configuration
    public class RedisConfig {
    
        @Bean
        public RedisTemplate<String, MessageEntity> redisTemplate(RedisConnectionFactory factory) {
            RedisTemplate<String, MessageEntity> template = new RedisTemplate<String, MessageEntity>();
            template.setConnectionFactory(factory);
    
            Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(
                    Object.class);
            ObjectMapper om = new ObjectMapper();
            om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
            om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
            jackson2JsonRedisSerializer.setObjectMapper(om);
            template.setValueSerializer(jackson2JsonRedisSerializer);
    
            template.setKeySerializer(new StringRedisSerializer());
            template.setHashKeySerializer(new StringRedisSerializer());
            template.afterPropertiesSet();
            return template;
        }
        
    }

用到的 redis 配置信息如下：

#数据库索引（默认为0）
    spring.redis.database=0
    spring.redis.host=localhost
    spring.redis.port=6379
    # Redis服务器连接密码（默认为空）
    spring.redis.password=
    #连接超时时间
    spring.redis.timeout=10000
    #最大连接数
    spring.redis.jedis.pool.max-active=8
    #最大阻塞等待时间(负数表示没限制)
    spring.redis.jedis.pool.max-wait=-1
    #最大空闲
    spring.redis.jedis.pool.max-idle=8
    #最小空闲
    spring.redis.jedis.pool.min-idle=0
    
    
    # redis消息队列键名
    redis.queue.key=queue
    # redis消息队列读取消息超时时间，单位：秒
    redis.queue.pop.timeout=1000

3.消息的消费者，消费者需要不断轮询队列，有消息便取出来。

import org.springframework.beans.factory.annotation.Value;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.Resource;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 消息的消费者，消费者需要不断的去轮询队列，有消息就取出来
     */
    @Service
    public class MessageConsumerService extends Thread {
    
    	@Resource
    	private RedisTemplate<String, MessageEntity> redisTemplate;
    
    	private volatile boolean flag = true;
    
    	@Value("${redis.queue.key}")
    	private String queueKey;
    
    	@Value("${redis.queue.pop.timeout}")
    	private Long popTimeout;
    
    	@Override
    	public void run() {
    		try {
    			MessageEntity message;
    			while (flag && !Thread.currentThread().isInterrupted()) {
    				message = redisTemplate.opsForList().rightPop(queueKey, popTimeout, TimeUnit.SECONDS);
    				System.out.println("接收到了" + message);
    			}
    		} catch (Exception e) {
    			System.err.println(e.getMessage());
    		}
    	}
    
    	public boolean isFlag() {
    		return flag;
    	}
    
    	public void setFlag(boolean flag) {
    		this.flag = flag;
    	}
    
    }

4.消息的生产者，这个类提供一个发送消息的方法。

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.stereotype.Service;
    
    /**
     * 生产者：提供一个发送消息的方法
     */
    @Service
    public class MessageProducerService {
    
    	@Autowired
    	private RedisTemplate<String, MessageEntity> redisTemplate;
    
    	@Value("${redis.queue.key}")
    	private String queueKey;
    
    	public Long sendMeassage(MessageEntity message) {
    		System.out.println("发送了" + message);
    		return redisTemplate.opsForList().leftPush(queueKey, message);
    	}
    
    }

## 测试 ##

import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class QueueTest {
    
    	@Autowired
    	private MessageProducerService producer;
    
    	@Autowired
    	private MessageConsumerService consumer;
    
    	/**
    	 * 消息队列模型测试
    	 */
    	@Test
    	public void testQueue() {
    		consumer.start();
    		producer.sendMeassage(new MessageEntity("1", "aaaa"));
    		producer.sendMeassage(new MessageEntity("2", "bbbb"));
    
    		try {
    			Thread.sleep(2000L);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		consumer.interrupt();
    	}
    }

输出信息如下：

2020-06-28 11:27:08.136  INFO 81195 --- [           main] DeferredRepositoryInitializationListener : Spring Data repositories initialized!
    2020-06-28 11:27:08.147  INFO 81195 --- [           main] c.test.springboot2_test.queue.QueueTest  : Started QueueTest in 11.012 seconds (JVM running for 12.666)
    发送了MessageEntity [id=1, content=aaaa]
    发送了MessageEntity [id=2, content=bbbb]
    接收到了MessageEntity [id=1, content=aaaa]
    接收到了MessageEntity [id=2, content=bbbb]
    Redis command interrupted; nested exception is io.lettuce.core.RedisCommandInterruptedException: Command interrupted

至此，消息队列的方式也整合完成了。  
虽然redisTemplate是线程安全的，但是如果一个队列有多个接收者的话，可能也还需要考虑一下并发的问题。

### 发布/订阅模型 ###

发布/订阅模型，因为Spring官方给了[示例][Link 2]。但是呢，示例里面的消息是String类型，对于我们的业务来说，可能更需要一个POJO（MessageEntity实体类）

1.先学习下org.springframework.data.redis.listener.adapter.MessageListenerAdapter源码如下，可以看到，如果使用StringRedisTemplate的话，默认都是使用StringRedisSerializer来反序列化，而如果想主动接收消息，则需要实现MessageListener接口。

/**
         * Standard Redis {@link MessageListener} entry point.
         * <p>
         * Delegates the message to the target listener method, with appropriate conversion of the message argument. In case
         * of an exception, the {@link #handleListenerException(Throwable)} method will be invoked.
         * 
         * @param message the incoming Redis message
         * @see #handleListenerException
         */
        public void onMessage(Message message, byte[] pattern) {
            try {
                // Check whether the delegate is a MessageListener impl itself.
                // In that case, the adapter will simply act as a pass-through.
                if (delegate != this) {
                    if (delegate instanceof MessageListener) {
                        ((MessageListener) delegate).onMessage(message, pattern);
                        return;
                    }
                }
    
                // Regular case: find a handler method reflectively.
                Object convertedMessage = extractMessage(message);
                String convertedChannel = stringSerializer.deserialize(pattern);
                // Invoke the handler method with appropriate arguments.
                Object[] listenerArguments = new Object[] { convertedMessage, convertedChannel };
    
                invokeListenerMethod(invoker.getMethodName(), listenerArguments);
            } catch (Throwable th) {
                handleListenerException(th);
            }
        }
    
        /**
         * Extract the message body from the given Redis message.
         * 
         * @param message the Redis <code>Message</code>
         * @return the content of the message, to be passed into the listener method as argument
         */
        protected Object extractMessage(Message message) {
            if (serializer != null) {
                return serializer.deserialize(message.getBody());
            }
            return message.getBody();
        }
    
        /**
         * Initialize the default implementations for the adapter's strategies.
         * 
         * @see #setSerializer(RedisSerializer)
         * @see JdkSerializationRedisSerializer
         */
        protected void initDefaultStrategies() {
            RedisSerializer<String> serializer = new StringRedisSerializer();
            setSerializer(serializer);
            setStringSerializer(serializer);
        }

2.spring data redis实现发布与订阅需要配置以下信息：

*  Topic
 *  MessageListener
 *  RedisMessageListenerContainer

1）用到的相关依赖：

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

2）配置 spring data redis:

import com.fasterxml.jackson.annotation.JsonAutoDetect;
    import com.fasterxml.jackson.annotation.PropertyAccessor;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.test.springboot2_test.queue.ConsumerRedisListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cache.CacheManager;
    import org.springframework.cache.annotation.CachingConfigurerSupport;
    import org.springframework.cache.annotation.EnableCaching;
    import org.springframework.cache.interceptor.KeyGenerator;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.cache.RedisCacheConfiguration;
    import org.springframework.data.redis.cache.RedisCacheManager;
    import org.springframework.data.redis.cache.RedisCacheWriter;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.data.redis.listener.ChannelTopic;
    import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
    import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
    import org.springframework.data.redis.serializer.RedisSerializationContext;
    import org.springframework.data.redis.serializer.StringRedisSerializer;
    
    import java.lang.reflect.Method;
    import java.time.Duration;
    
    /**
     * Redis 缓存配置类
     */
    @Configuration
    @EnableCaching
    public class RedisConfig extends CachingConfigurerSupport {
    
    	@Bean
    	public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
    		RedisTemplate<String, Object> template = new RedisTemplate<>();
    		template.setConnectionFactory(factory);
    
    		Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(
    				Object.class);
    		ObjectMapper om = new ObjectMapper();
    		om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
    		om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
    		jackson2JsonRedisSerializer.setObjectMapper(om);
    		template.setValueSerializer(jackson2JsonRedisSerializer);
    
    		template.setKeySerializer(new StringRedisSerializer());
    		template.setHashKeySerializer(new StringRedisSerializer());
    		template.afterPropertiesSet();
    		return template;
    	}
    
    	@Autowired
    	private LettuceConnectionFactory connectionFactory;
    
    	@Bean
    	public ConsumerRedisListener consumeRedis() {
    		return new ConsumerRedisListener();
    	}
    
    	@Bean
    	public ChannelTopic topic() {
    		return new ChannelTopic("topic");
    	}
    
    	@Bean
    	public RedisMessageListenerContainer redisMessageListenerContainer() {
    		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    		container.setConnectionFactory(connectionFactory);
    		container.addMessageListener(consumeRedis(), topic());
    		return container;
    	}
    
    }

3）实现一个Object类型的 topic MessageListener

import org.springframework.data.redis.connection.Message;
    import org.springframework.data.redis.connection.MessageListener;
    import org.springframework.data.redis.core.RedisTemplate;
    
    import javax.annotation.Resource;
    
    public class ConsumerRedisListener implements MessageListener {
    
    	@Resource
    	private RedisTemplate<String, Object> redisTemplate;
    
    	@Override
    	public void onMessage(Message message, byte[] pattern) {
    		doBusiness(message);
    	}
    
    	/**
    	 * 打印 message body 内容
    	 * @param message
    	 */
    	public void doBusiness(Message message) {
    		Object value = redisTemplate.getValueSerializer().deserialize(message.getBody());
    		System.out.println("consumer message: " + value.toString());
    	}
    
    }

4）其他（最简单的application.properties配置如下）

通过上面四步，简单的订阅者就做好了，通过以下代码可以发布一个消息，同时可以查看到控制台会有订阅者消费信息打印出来：

import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import javax.annotation.Resource;
    import java.util.Date;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class QueueTest {
    
    	@Resource
    	private RedisTemplate<String, Object> redisTemplate;
    
    	/**
    	 * 发布订阅模型测试
    	 */
    	@Test
    	public void testSubscribe() {
    		String channel = "topic";
    
    		System.out.println("开始发送消息。。。。。。。。。。。。。。");
    		redisTemplate.convertAndSend(channel, "hello world");
    		redisTemplate.convertAndSend(channel, new Date(System.currentTimeMillis()));
    		redisTemplate.convertAndSend(channel, new MessageEntity("1", "object"));
    		System.out.println("发送消息结束。。。。。。。。。。。。。。");
    
    	}
    
    }

这里用到了一个实体类用于测试。

import java.io.Serializable;
    
    public class MessageEntity implements Serializable {
    	private static final long serialVersionUID = 8632296967087444509L;
    
    	private String id;
    
    	private String content;
    
    	public MessageEntity() {
    		super();
    	}
    
    	public MessageEntity(String id, String content) {
    		super();
    		this.id = id;
    		this.content = content;
    	}
    
    	public static long getSerialVersionUID() {
    		return serialVersionUID;
    	}
    
    	public String getId() {
    		return id;
    	}
    
    	public void setId(String id) {
    		this.id = id;
    	}
    
    	public String getContent() {
    		return content;
    	}
    
    	public void setContent(String content) {
    		this.content = content;
    	}
    
    	@Override
    	public String toString() {
    		return "MessageEntity [id=" + id + ", content=" + content + "]";
    	}
    
    }

输出结果如下：

consumer message: hello world
    consumer message: Sat Feb 23 13:04:40 CST 2019
    consumer message: MessageEntity [id=1, content=object]

用 spring data redis 来实现 redis 订阅者，本质上还是Listener模式，只需要配置Topic, MessageListener 和 RedisMessageListenerContainer就可以了。同时，发布时，只需要使用 redisTemplate 的 convertAndSend方法即可topic来发布message。

参考：

[springboot整合redis消息队列][springboot_redis]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h4YzE2MDU2Mjk4OTU_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190223132431644.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h4YzE2MDU2Mjk4OTU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h4YzE2MDU2Mjk4OTU_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/201902231328581.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h4YzE2MDU2Mjk4OTU=,size_16,color_FFFFFF,t_70
[Link 1]: https://link.jianshu.com/?t=https%3A%2F%2Fdocs.spring.io%2Fspring-data%2Fdata-redis%2Fdocs%2Fcurrent%2Fapi%2Forg%2Fspringframework%2Fdata%2Fredis%2Fcore%2FRedisTemplate.html
[Link 2]: https://link.jianshu.com/?t=https%3A%2F%2Fspring.io%2Fguides%2Fgs%2Fmessaging-redis%2F
[springboot_redis]: https://blog.csdn.net/johnf_nash/article/details/87891293