KeyExpirationEventMessageListener Key过期监听事件,服务集群出现的业务问题

港控/mmm° 2024-02-19 21:50 42阅读 0赞

**项目有个业务场景,到某个时间节点时,固定向用户的公众号发送消息通知~**

**因为时间点不固定所以没有采用定时任务，  
使用redis的key失效监听器来做，思路就是将key保存的时候，  
算好当前和那个需要发送通知时候的时间间隔作为key失效时间，这样就可以保证到点实时发送消息了。**

**单机模式下测试推送正常，但是到线上的时候，用户收到了两条推送，原因是因为服务开了集群，key失效的时候每个服务都收到了通知，这时候进行消息的推送，所以发生了推送多条消息的问题。**

**这时候可以考虑采用redis的setNx命令实现锁竞争，同一时间内只有一个服务能对同一个key进行抢占使用，也就是发送消息，其他没有获取到锁的服务，则直接放弃执行消息推送。**

**setNx命令原理就是如果key不存在的话，保存成功就返回 1 否则就返回 0。  
redis的命令是单线程执行的，所以同一时间段内肯定只能有一个人能拿到成功的结果。**

**实现代码如下**

@Component
    public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
    
        @Autowired
        private RedisRepository redisRepository;
    
        public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer){
            super(listenerContainer);
        }
    
        @Override
        public void onMessage(Message message, byte[] pattern) {
            String expiredKey = message.toString();
            // 获取锁 重点，此处就是多个服务或者线程对同一个key进行抢占的setNx命令调用
            if(redisRepository.lockBySecondsTime(key, 30)){
                // 消息通知伪代码
    			push();
            }
            super.onMessage(message, pattern);
        }
    
    }

lockBySecondsTime 锁方法实现逻辑如下

public boolean lockBySecondsTime(String key, long expirationTime){
            //  此方法只适用于在 expirationTime 时间段内进行锁竞争的场景。如果超过 expirationTime 时间段，锁自动失效，之前获取到锁的线程还在运行，就失去了分布式锁的意义，慎重根据自己的场景来使用。
            Long timeStamp = new Date().getTime() + (expirationTime * 1000);
            // 通过setNx获取锁
            return ifAbsent(key, String.valueOf(timeStamp), expirationTime, TimeUnit.SECONDS);
        }
    
    	public boolean ifAbsent(String key, String value, long expirationTime , TimeUnit timeUnit) {
            Boolean res = (Boolean) redisTemplate.execute(new RedisCallback() {
                @Override
                public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
                    return connection.stringCommands().set(key.getBytes(), value.getBytes(),
                            Expiration.from(expirationTime, timeUnit), RedisStringCommands.SetOption.ifAbsent());
                }
            });
            return res == null ? false : res;
        }

**以上的代码是依赖SpringDataRedis模块的，关注一下ifAbsent方法，此处使用了 redisTemplate.execute方法，为什么要这么做？**

**如果你是2.1.0版本或者更高的版本 可以直接通过 redisTemplate.opsForValue().setIfAbsent(key,value,time)的方式完成 上面代码中的 ifAbsent操作，**

**但是2.1.0以下的版本只有redisTemplate.opsForValue().setIfAbsent(key,value) 方法，缺少了一个设置key过期时间，如果使用这个方法的话，key就会一直存在redis当中，造成内存空间资源浪费，所以此时笔者采用redisTemplate.execute拓展了一下，保证这个key作为锁竞争资源以后，后面会自行消失。**

**采用redis锁的方式，使得同一时间内对同一个key操作只能由一个服务来做，所以这时候我的消息通知也就正常啦，不会有重复的消息推送~**

**推荐一个基于redis的分布式锁框架 redisson 有兴趣可以了解一下~**