本篇内容介绍了“Redis监听过期的key实现流程是什么”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！一、简介我们来个最简单的集群架构，如下
本篇内容介绍了“Redis监听过期的key实现流程是什么”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
我们上面图中看到是服务A和服务B就是同一个服务的不同实例。
pom.xml
<?xml version="1.0" encoding="UTF-8"?><project xmlns="Http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.6.0</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>com.alian</groupId> <artifactId>expiration</artifactId> <version>0.0.1-SNAPSHOT</version> <name>expiration</name> <description>Redis-key-expiration-listener</description> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <project.package.directory>target</project.package.directory> <java.version>1.8</java.version> <!--com.fasterxml.jackson 版本--> <jackson.version>2.9.10</jackson.version> <!--阿里巴巴fastJSON 版本--> <fastjson.version>1.2.68</fastjson.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-WEB</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> </dependency> <!--redis依赖--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> <version>${parent.version}</version> </dependency> <!--用于序列化--> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>${jackson.version}</version> </dependency> <!--java 8时间序列化--> <dependency> <groupId>com.fasterxml.jackson.datatype</groupId> <artifactId>jackson-datatype-jsr310</artifactId> <version>${jackson.version}</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.68</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.16.14</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.13.2</version> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build></project># 端口
server.port=8090
# 上下文路径
server.servlet.context-path=/expiration# Redis数据库索引（默认为0）
spring.redis.database=0
# Redis服务器地址
spring.redis.host=192.168.0.193
#spring.redis.host=127.0.0.1
# Redis服务器连接端口
spring.redis.port=6379
# Redis服务器连接密码（默认为空）
spring.redis.passWord=
# 连接池最大连接数（使用负值表示没有限制）
spring.redis.jedis.pool.max-active=20
# 连接池中的最小空闲连接
spring.redis.jedis.pool.min-idle=10
# 连接池中的最大空闲连接
spring.redis.jedis.pool.max-idle=10
# 连接池最大阻塞等待时间（使用负值表示没有限制）
spring.redis.jedis.pool.max-wait=20000
# 读时间（毫秒）
spring.redis.timeout=10000
# 连接超时时间（毫秒）
spring.redis.connect-timeout=10000
RedisConfig
package com.alian.expiration.config;import com.fasterxml.jackson.annotation.JsonAutoDetect;import com.fasterxml.jackson.annotation.PropertyAccessor;import com.fasterxml.jackson.databind.ObjectMapper;import com.fasterxml.jackson.databind.SerializationFeature;import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;import com.fasterxml.jackson.datatype.jsr310.deser.LocalDateDeserializer;import com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer;import com.fasterxml.jackson.datatype.jsr310.deser.LocalTimeDeserializer;import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateSerializer;import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateTimeSerializer;import com.fasterxml.jackson.datatype.jsr310.ser.LocalTimeSerializer;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.listener.RedisMessageListenerContainer;import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;import org.springframework.data.redis.serializer.RedisSerializer;import org.springframework.data.redis.serializer.StringRedisSerializer;import java.time.LocalDate;import java.time.LocalDateTime;import java.time.LocalTime;import java.time.fORMat.DateTimeFormatter;@Configurationpublic class RedisConfig { @Bean public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) { // 实例化redisTemplate RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>(); //设置连接工厂 redisTemplate.setConnectionFactory(redisConnectionFactory); // key采用String的序列化 redisTemplate.seTKEySerializer(keySerializer()); // value采用jackson序列化 redisTemplate.setValueSerializer(valueSerializer()); // Hash key采用String的序列化 redisTemplate.setHashKeySerializer(keySerializer()); // Hash value采用jackson序列化 redisTemplate.setHashValueSerializer(valueSerializer()); // 支持事务 // redisTemplate.setEnableTransactionSupport(true); //执行函数，初始化RedisTemplate redisTemplate.afterPropertiesSet(); return redisTemplate; } private RedisSerializer<String> keySerializer() { return new StringRedisSerializer(); } private RedisSerializer<Object> valueSerializer() { //设置jackson序列化 Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class); //设置序列化对象 jackson2JsonRedisSerializer.setObjectMapper(getMapper()); return jackson2JsonRedisSerializer; } private ObjectMapper getMapper() { ObjectMapper mapper = new ObjectMapper(); //设置可见性 mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY); //默认键入对象 mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL); //设置Java 8 时间序列化 JavaTimeModule timeModule = new JavaTimeModule(); timeModule.addSerializer(LocalDateTime.class, new LocalDateTimeSerializer(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))); timeModule.addSerializer(LocalDate.class, new LocalDateSerializer(DateTimeFormatter.ofPattern("yyyy-MM-dd"))); timeModule.addSerializer(LocalTime.class, new LocalTimeSerializer(DateTimeFormatter.ofPattern("HH:mm:ss"))); timeModule.aDDDeserializer(LocalDateTime.class, new LocalDateTimeDeserializer(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))); timeModule.addDeserializer(LocalDate.class, new LocalDateDeserializer(DateTimeFormatter.ofPattern("yyyy-MM-dd"))); timeModule.addDeserializer(LocalTime.class, new LocalTimeDeserializer(DateTimeFormatter.ofPattern("HH:mm:ss"))); //禁用把时间转为时间戳 mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false); mapper.reGISterModule(timeModule); return mapper; } @Bean RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); return container; }}和我们之前整合redis差不多，只不过在最后增加了一个redis消息监听监听容器RedisMessageListenerContainer
RedisKeyExpirationListener
package com.alian.expiration.listener;import com.alian.expiration.service.RedisExpirationService;import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.redis.connection.Message;import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;import org.springframework.data.redis.listener.RedisMessageListenerContainer;import org.springframework.stereotype.Component;@Slf4j@Componentpublic class RedisKeyExpirationListener extends KeyExpirationEventMessageListener { @Autowired private RedisExpirationService redisExpirationService;// 把我们上面一步配置的bean注入进去 public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) { super(listenerContainer); } @Override public void onMessage(Message message, byte[] pattern) { // 用户做自己的业务处理即可,注意message.toString()可以获取失效的key String expiredKey = message.toString(); log.info("onMessage --> redis 过期的key是：{}", expiredKey); try { // 对过期key进行处理 redisExpirationService.processingExpiredKey(expiredKey); log.info("过期key处理完成：{}", expiredKey); } catch (Exception e) { e.printStackTrace(); log.error("处理redis 过期的key异常：{}", expiredKey, e); } }}实现的步骤如下：
继承KeyExpirationEventMessageListener
把redis消息监听监听容器RedisMessageListenerContainer 注入到密钥空间事件消息侦 听器中
重写onMessage方法
通过Message 的 toString() 方法就可以获取到过期的key
对key中关键信息进行业务处理，比如 id
RedisExpirationService
package com.alian.expiration.service;import com.alian.expiration.util.SignUtils;import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.redis.core.RedisTemplate;import org.springframework.stereotype.Service;import java.util.concurrent.TimeUnit;@Slf4j@Servicepublic class RedisExpirationService { @Autowired private RedisTemplate<String, Object> redisTemplate; public void processingExpiredKey(String expiredKey) { // 如果是优惠券的key(一定要规范命名) if (expiredKey.startsWith("com.mall.coupon.id")) { // 临时key,此key可以在业务处理完，然后延迟一定时间删除，或者不处理 String tempKey = SignUtils.md5(expiredKey, "UTF-8"); // 临时key不存在才设置值，key超时时间为10秒（此处相当于分布式锁的应用） Boolean exist = redisTemplate.opsForValue().setIfAbsent(tempKey, "1", 10, TimeUnit.SECONDS); if (Boolean.TRUE.equals(exist)) { log.info("Business Handing..."); // 比如截取里面的id，然后关联数据库进行处理 } else { log.info("Other service is handing..."); } } else { log.info("Expired keys without processing"); } }}基本流程如下：
判断是否是需要处理的key，一般这种key通过命名规范加以处理
以当前key生成一个新的key作为分布式key
如果redis中不存在这个新的key，则为新的key设置一个值，达到分布式服务处理（核心）
设置成功的，进行业务处理；设置失败了，说明其他服务正在处理这个key
根据 key 的关键信息（比如截取id），进行业务处理
SignUtils
package com.alian.expiration.util;import java.security.MessageDigest;public class SignUtils { public static final String md5(String s, String charset) { char[] hexDigits = new char[]{'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F'}; try { byte[] btInput = s.getBytes(charset); MessageDigest mdInst = MessageDigest.getInstance("MD5"); mdInst.update(btInput); byte[] md = mdInst.digest(); int j = md.length; char[] str = new char[j * 2]; int k = 0; for (byte byte0 : md) { str[k++] = hexDigits[byte0 >>> 4 & 15]; str[k++] = hexDigits[byte0 & 15]; } return new String(str); } catch (Exception var11) { return ""; } }}简单模拟下发送一个优惠券数据到redis，然后设置超时时间
package com.alian.expiration;import lombok.extern.slf4j.Slf4j;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.data.redis.core.RedisTemplate;import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;import java.util.HashMap;import java.util.Map;import java.util.concurrent.TimeUnit;@Slf4j@RunWith(SpringJUnit4ClassRunner.class)@SpringBootTestpublic class RedisKeyExpirationTest { @Autowired private RedisTemplate<String, Object> redisTemplate; @Test public void keyExpiration() { // 优惠券信息 String id = "2023021685264735"; Map<String, String> map = new HashMap<>(); map.put("id", id); map.put("amount", "1000"); map.put("type", "1001"); map.put("describe", "满减红包"); // 缓存到redis redisTemplate.opsForHash().putAll("com.mall.coupon.id." + id, map); // 设置过期时间 redisTemplate.expire("com.mall.coupon.id." + id, 10, TimeUnit.SECONDS); }}单实例就是服务只部署了一份，我们启动一份，端口是8090，然后通过上面的测试类，发送一个消息，结果如下：
39 701 INFO [container-2]:onMessage --> redis 过期的key是：com.mall.coupon.id.2023021685264735
10:23:39 988 INFO [container-2]:Business Handing...
10:23:39 989 INFO [container-2]:过期key处理完成：com.mall.coupon.id.2023021685264735
10:23:50 005 INFO [container-3]:onMessage --> redis 过期的key是：450FCC35415BADC16805962CA5BC7E12
10:23:50 005 INFO [container-3]:Expired keys without processing
10:23:50 005 INFO [container-3]:过期key处理完成：450FCC35415BADC16805962CA5BC7E12
多实例就是服务部署了多份，比如我们启动两份，端口分别为8090和8091，然后通过上面的测试类，发送一个消息，8090端口的服务结果如下（Business Handing…）：
06 691 INFO [container-2]:onMessage --> redis 过期的key是：com.mall.coupon.id.2023021685264735
11:39:06 707 INFO [container-2]:Business Handing...
11:39:06 707 INFO [container-2]:过期key处理完成：com.mall.coupon.id.2023021685264735
11:39:16 796 INFO [container-3]:onMessage --> redis 过期的key是：450FCC35415BADC16805962CA5BC7E12
11:39:16 796 INFO [container-3]:Expired keys without processing
11:39:16 796 INFO [container-3]:过期key处理完成：450FCC35415BADC16805962CA5BC7E12
8091端口的服务结果如下（Other service is handing…）：
06 691 INFO [container-2]:onMessage --> redis 过期的key是：com.mall.coupon.id.2023021685264735
11:39:06 707 INFO [container-2]:Other service is handing...
11:39:06 707 INFO [container-2]:过期key处理完成：com.mall.coupon.id.2023021685264735
11:39:16 796 INFO [container-3]:onMessage --> redis 过期的key是：450FCC35415BADC16805962CA5BC7E12
11:39:16 796 INFO [container-3]:Expired keys without processing
11:39:16 796 INFO [container-3]:过期key处理完成：450FCC35415BADC16805962CA5BC7E12
结果分析：
多实例的情况下，每个实例都会收到过期key通知
通过redis分布式锁，实现只有一个实例会进行业务处理，防止重复
使用分布式锁会有一个新的key过期，并且收到该key的通知，你可以业务执行完延迟一定时间（避免重复执行），再删除，也可以不处理（因为本就不是要处理业务的key）
“Redis监听过期的key实现流程是什么”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注编程网网站，小编将为大家输出更多高质量的实用文章！
--结束END--
本文标题: Redis监听过期的key实现流程是什么
本文链接: https://lsjlt.com/news/350252.html(转载时请注明来源链接)
有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
回答
回答
回答
回答
回答
回答
回答
回答
回答
回答
官方手机版
微信公众号
商务合作
0