SpringBoot整合Redis，订阅、发布、过期事件

作者: 程就人生 | 来源:发表于2020-03-15 22:43 被阅读0次

Redis可以实现消息中间件的功能，在实现这个功能时，又有几种实现方式呢，看了网友们的分享，做了一个总结；消息的订阅有两种形式，消息的发布也有两种形式，今天就来说说有哪两种形式，另外把Redis过期事件的监听也整理整理。

消息的订阅有两种方式，一种是在Redis配置文件中订阅，另一种是使用连接工厂类在需要的任意时候订阅；消息的发布也有两种，一种是使用连接工厂类进行发布，另一种是使用RedisTemplate模板发布，下面看代码。

首先，在pom中引入springboot和redis的集成架包；

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

第二，在properties文件里加入redis的配置；

spring.redis.database=0
spring.redis.host=127.0.0.1
spring.redis.port=6379
spring.redis.password=
spring.redis.timeout=3000
spring.redis.lettuce.pool.max-idle=100
spring.redis.lettuce.pool.min-idle=50
spring.redis.lettuce.pool.max-wait=2000

第三，redis的配置文件；

import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import com.example.demo.subscribe.MessageReceiveOne;
import com.example.demo.subscribe.MessageReceiveTwo;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * redis通用配置文件
 * @author 程就人生
 * @date 2020年2月24日
 */
@Configuration
@AutoConfigureAfter(RedisAutoConfiguration.class)
public class RedisConfig{
    
    /**
     * RedisTemplate模板bean的初始化
     * @param factory
     * @return
     */
    @SuppressWarnings({ "rawtypes", "unchecked"})
    @Bean("redisTemplate")
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory factory) {
        
        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
        
        template.setConnectionFactory(factory);
        
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        
        ObjectMapper om = new ObjectMapper();
        
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        
        jackson2JsonRedisSerializer.setObjectMapper(om);
        
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        
        // key采用String的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        
        // hash的key也采用String的序列化方式
        template.setHashKeySerializer(stringRedisSerializer);
        
        // value序列化方式采用jackson
        template.setValueSerializer(jackson2JsonRedisSerializer);
        
        // hash的value序列化方式采用jackson
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        
        template.afterPropertiesSet();
         
        return template;
    }
    
    /**
     * redis消息监听器容器
     * 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
     * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
     * @param connectionFactory
     * @param listenerAdapter
     * @return
     */
    //MessageListenerAdapter 表示监听频道的不同订阅者
    @SuppressWarnings({ "rawtypes", "unchecked" })
    @Bean
    RedisMessageListenerContainer container(LettuceConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter2,MessageListenerAdapter listenerAdapter){
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //订阅一个频道
        container.addMessageListener(listenerAdapter, new PatternTopic("one"));
        //订阅多个频道
        container.addMessageListener(listenerAdapter2, new PatternTopic("one"));
        container.addMessageListener(listenerAdapter2, new PatternTopic("two"));

        //序列化对象（特别注意：发布的时候需要设置序列化；订阅方也需要设置序列化）
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        
        ObjectMapper objectMapper = new ObjectMapper();
        
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        
        container.setTopicSerializer(jackson2JsonRedisSerializer);
        
        return container;
    }
    
    /**
     * 过期时间监听
     * @param factory
     * @return
     */
    @Bean
    public RedisMessageListenerContainer container2(LettuceConnectionFactory factory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.addMessageListener(new MessageListener(){
            @Override
            public void onMessage(Message message, byte[] pattern) {
                System.out.println(message.toString());
                System.out.println(new String(pattern));
            }
            
        }, new PatternTopic("__keyevent@0__:expired"));//__key*__:*
        return container;
    }

    /**
     * 第一个监听频道
     * @param receiver
     * @return
     */
    @Bean
    public MessageListenerAdapter listenerAdapter(MessageReceiveOne receiver){
        //这个地方 是给messageListenerAdapter 传入一个消息接受的处理器，利用反射的方法调用“MessageReceiveOne ”
        return new MessageListenerAdapter(receiver, "getMessage");
    }    
    
    /**
     * 第二个监听频道
     * @param receiver
     * @return
     */
    @Bean
    public MessageListenerAdapter listenerAdapter2(MessageReceiveTwo receiver){
        return new MessageListenerAdapter(receiver, "getMessage");
    }    
}

第四，消息监听者的实现；

import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * 第一个消费者
 * @author 程就人生
 * @Date
 */
@Component
public class MessageReceiveOne {
    
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public void getMessage(String object){
        
        //序列化对象，发布、订阅序列号保持一致
        Jackson2JsonRedisSerializer seria = new Jackson2JsonRedisSerializer(User.class);
        
        ObjectMapper objectMapper = new ObjectMapper();
        
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        
        seria.setObjectMapper(objectMapper);
        
        User user = (User)seria.deserialize(object.getBytes());
        
        System.out.println("消息客户端1号："+ user.toString());
    }
}

import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * 第二个消费者
 * @author 程就人生
 * @Date
 */
@Component
public class MessageReceiveTwo {
    
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public void getMessage(String object){
        
        //序列化对象，订阅发布序列号化保持一致
        Jackson2JsonRedisSerializer seria = new Jackson2JsonRedisSerializer(User.class);
        
        ObjectMapper objectMapper = new ObjectMapper();
        
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        
        seria.setObjectMapper(objectMapper);
        
        User user = (User)seria.deserialize(object.getBytes());
        
        System.out.println("消息客户端2号：" + user.toString());
    }
}

/**
 * 第三个消费者
 * @author 程就人生
 * @Date
 */
public class MessageReceiveThree implements MessageListener{
    
    @Override
    public void onMessage(Message message, byte[] pattern) {
        System.out.println("消息客户端3号："+ message.toString());
    }
}

第四，测试代码

import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import com.example.demo.subscribe.MessageReceiveThree;
import com.example.demo.subscribe.User;

/**
 * 测试程序
 * @author FengJuan
 * @Date
 */
@RestController
public class TestController {

    @Autowired
    private RedisTemplate<String,Object> redisTemplate;
    
    @Autowired
    private LettuceConnectionFactory factory;
    
    /**
     * 在配置文件里监听两个事件
     */
    @GetMapping("/index")
    public void index(){
        String channel1 = "one";
        String channel2 = "two";
        User user = new User();
        user.setPhone("18988888888");
        user.setName("张三");

        User user2 = new User();
        user2.setPhone("18988888889");
        user2.setName("李四");

        //发布消息
        redisTemplate.convertAndSend(channel1,user2);       
        redisTemplate.convertAndSend(channel2,user);        
    }
    
    
    /**
     * 不再配置文件里的监听
     */
    @GetMapping("/index1")
    public void index1(){
        //随时监听
        factory.getConnection().subscribe(new MessageReceiveThree(), "three".getBytes());
        
        User user2 = new User();
        user2.setPhone("18988888889");
        user2.setName("王五");
        
        factory.getConnection().publish("three".getBytes(), user2.toString().getBytes());
    }
    
    
    /**
     * 过期事件的监听
     */
    @GetMapping("/index2")
    public void index2(){
        
        redisTemplate.opsForHash().put("aaakey", "bbbb", "cccc");
        //设置该key五秒后过期
        redisTemplate.expire("aaakey", 5, TimeUnit.SECONDS);
    }
}

备注：
Redis的过期时间，需要在配置文件redis.windows.conf中加入这样一句配置，然后重启，过期事件才能生效；

notify-keyspace-events Ex

这个Ex是什么意思呢，x指过期时间，E指keyevent事件，Ex则是keyevent的过期事件，除了这个事件，还有其他的一些事件，如下：

K：keyspace事件，事件以__keyspace@<db>__为前缀进行发布；         
E：keyevent事件，事件以__keyevent@<db>__为前缀进行发布；         
g：一般性的，非特定类型的命令，比如del，expire，rename等；        
$：字符串特定命令；         
l：列表特定命令；         
s：集合特定命令；         
h：哈希特定命令；         
z：有序集合特定命令；         
x：过期事件，当某个键过期并删除时会产生该事件；         
e：驱逐事件，当某个键因maxmemore策略而被删除时，产生该事件；         
A：g$lshzxe的别名，因此”AKE”意味着所有事件。

最后，启动测试，在浏览器地址栏一次输入请求url地址，然后查看控制台输出；

测试结果

总结
发布和订阅的写法看似不同，仔细琢磨一下，又没有什么不同，只是在写法上变通了一下，具体使用哪种写法，还需要看具体的业务需求。

参考文档：
https://www.cnblogs.com/liuyp-ken/p/10538658.html
https://www.jianshu.com/p/6ed351954c21
https://blog.csdn.net/weixin_33979203/article/details/91944365

SpringBoot下，Redis单例和集成的通用配置
 Windows安装Redis集群，并整合SpringBoot2
Redis支持的五种数据类型操作，详细整理
 SpringBoot2.1 Redis 发布/订阅模式
 SpringBoot2.1 Redis多数据源集成
 SpringBoot初次集成Redis踩坑，安装+使用

SpringBoot整合Redis，订阅、发布、过期事件

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

SpringBoot精选