Of Stack

Implementation code for expired key listening under Redis cluster

  • 2020-06-23 02:14:40
  • OfStack

1. Introduction

When using the redis cluster, it was found that expired key was never listened for. There are no off-the-shelf solutions online. So I thought, since I can't listen to the cluster, I can set up multiple redis connections to listen for each redis's key expiration. The above approach may not be satisfactory, and no good solution has been found at present. If you have any good ideas, please leave a message! Not to mention, directly post my own code!

2. Code implementation

About Redis cluster configuration code not posted here, directly posted configuration monitoring class code! 


redis.host1: 10.113.56.68
redis.port1: 7030
redis.host2: 10.113.56.68
redis.port2: 7031
redis.host3: 10.113.56.68
redis.port3: 7032
redis.host4: 10.113.56.68
redis.port4: 7033
redis.host5: 10.113.56.68
redis.port5: 7034
redis.host6: 10.113.56.68
redis.port6: 7035

import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.CacheManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;

import java.util.Arrays;

/**
 * @Author xiabing5
 * @Create 2019/8/6 14:46
 * @Desc   Listening to the redis In the Key Expired events 
 **/
@Configuration
public class RedisListenerConfig {

  @Value("${redis.host1}")
  private String host1;

  @Value("${redis.host2}")
  private String host2;

  @Value("${redis.host3}")
  private String host3;

  @Value("${redis.host4}")
  private String host4;

  @Value("${redis.host5}")
  private String host5;

  @Value("${redis.host6}")
  private String host6;

  @Value("${redis.port1}")
  private int port1;

  @Value("${redis.port2}")
  private int port2;

  @Value("${redis.port3}")
  private int port3;

  @Value("${redis.port4}")
  private int port4;

  @Value("${redis.port5}")
  private int port5;

  @Value("${redis.port6}")
  private int port6;

  @Bean
  JedisPoolConfig jedisPoolConfig(){
    JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
    jedisPoolConfig.setMaxIdle(100);
    jedisPoolConfig.setMaxWaitMillis(1000);
    return jedisPoolConfig;
  }

  // redis-cluster Does not support key Expired listening, establishing multiple connections for each redis Node listening 
  @Bean
  RedisMessageListenerContainer redisContainer1() {
    final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
    jedisConnectionFactory.setHostName(host1);
    jedisConnectionFactory.setPort(port1);
    jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
    jedisConnectionFactory.afterPropertiesSet();
    container.setConnectionFactory(jedisConnectionFactory);
    return container;
  }

  @Bean
  RedisMessageListenerContainer redisContainer2() {
    final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
    jedisConnectionFactory.setHostName(host2);
    jedisConnectionFactory.setPort(port2);
    jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
    jedisConnectionFactory.afterPropertiesSet();
    container.setConnectionFactory(jedisConnectionFactory);
    return container;
  }

  @Bean
  RedisMessageListenerContainer redisContainer3() {
    final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
    jedisConnectionFactory.setHostName(host3);
    jedisConnectionFactory.setPort(port3);
    jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
    jedisConnectionFactory.afterPropertiesSet();
    container.setConnectionFactory(jedisConnectionFactory);
    return container;
  }

  @Bean
  RedisMessageListenerContainer redisContainer4() {
    final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
    jedisConnectionFactory.setHostName(host4);
    jedisConnectionFactory.setPort(port4);
    jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
    jedisConnectionFactory.afterPropertiesSet();
    container.setConnectionFactory(jedisConnectionFactory);
    return container;
  }

  @Bean
  RedisMessageListenerContainer redisContainer5() {
    final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
    jedisConnectionFactory.setHostName(host5);
    jedisConnectionFactory.setPort(port5);
    jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
    jedisConnectionFactory.afterPropertiesSet();
    container.setConnectionFactory(jedisConnectionFactory);
    return container;
  }

  @Bean
  RedisMessageListenerContainer redisContainer6() {
    final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
    jedisConnectionFactory.setHostName(host6);
    jedisConnectionFactory.setPort(port6);
    jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
    jedisConnectionFactory.afterPropertiesSet();
    container.setConnectionFactory(jedisConnectionFactory);
    return container;
  }

  @Bean
  RedisKeyExpirationListener redisKeyExpirationListener1() {
    return new RedisKeyExpirationListener(redisContainer1());
  }

  @Bean
  RedisKeyExpirationListener redisKeyExpirationListener2() {
    return new RedisKeyExpirationListener(redisContainer2());
  }

  @Bean
  RedisKeyExpirationListener redisKeyExpirationListener3() {
    return new RedisKeyExpirationListener(redisContainer3());
  }

  @Bean
  RedisKeyExpirationListener redisKeyExpirationListener4() {
    return new RedisKeyExpirationListener(redisContainer4());
  }

  @Bean
  RedisKeyExpirationListener redisKeyExpirationListener5() {
    return new RedisKeyExpirationListener(redisContainer5());
  }

  @Bean
  RedisKeyExpirationListener redisKeyExpirationListener6() {
    return new RedisKeyExpirationListener(redisContainer6());
  }

}

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 java.util.Date;

/**
 * @Author xiabing5
 * @Create 2019/9/4 9:47
 * @Desc  redis Overdue listening 
 **/
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
  @Autowired
  RedisUtil redisUtil;
  @Autowired
  LoginUserStatisticsMapper loginUserStatisticsMapper;
  public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
    super(listenerContainer);
  }
  @Override
  public void onMessage(Message message, byte[] pattern) {
    //  Users can do their own business processing ,message.toString() Invalid ones can be obtained key
    String mesg = message.toString();   
  }
}

3.Redis prevents expired key repeat listening

In the case of project cluster, after deploying multiple services, it is easy for redis expiration to be monitored by multiple services simultaneously to perform the same business logic, which is not what we expect. The synchronize keyword can be used for synchronization of methods under stand-alone deployment. But under clustering, distributed locking is used. Where there is a need to lock, just lock and unlock. To write Redis, post an redis distributed lock for your own use. 


import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import java.util.Collections;
import java.util.UUID;
/**
 * @Author xiabing5
 * @Create 2019/9/6 15:54
 * @Desc  redis A distributed lock 
 **/
@Component
public class RedisLock {
  @Autowired
  Jedis jedis;
  private static final String SET_IF_NOT_EXIST = "NX"; // NX If it doesn't exist key Is set value
  private static final String SET_WITH_EXPIRE_TIME = "PX"; // PX Said ms 
  //  lock 
  public String tryLock(String key,Long acquireTimeout) {
    //  Generate random value
    String identifierValue = UUID.randomUUID().toString();
    //  Set the timeout 
    Long endTime = System.currentTimeMillis() + acquireTimeout;
    //  Loop acquisition lock 
    while (System.currentTimeMillis() < endTime) {
      String result = jedis.set(key,identifierValue, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, acquireTimeout);
      if("OK".equals(result)) {
        return identifierValue;
      }
    }
    return null;
  }
  //  unlock 
//  public void delLock(String key,String identifierValue) {
//    //  Whether it's the same 1 The lock 
//    try{
//      if(jedis.get(key).equals(identifierValue)){
//        //  Operating on nonatomicity here can easily result in releasing locks that are not of their own 
//        jedis.del(key);
//      }
//    }catch(Exception e) {
//      e.printStackTrace();
//    }
//  }
  //  use Lua The unlock code 
  public void delLock(String key,String identifierValue) {
    try{
      String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
      Long result = (Long) jedis.eval(script, Collections.singletonList(key), Collections.singletonList(identifierValue));
      if (1 == result) {
        System.out.println(result+" Release the lock successfully ");
      } if (0 == result) {
        System.out.println(result+" Release lock failed ");
      }
    }catch (Exception e) {
      e.printStackTrace();
    }
  }
}

4. To summarize

Self-realization of a small demo, less nonsense. Small white write their own configuration class, understand the problem please leave a message! The realization of their own scheme felt not appropriate, just the basic needs to complete, but also continue to study.

Related articles: