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踩坑,安装+使用