Análisis superficial de la llave distribuida Redis

En los últimos tiempos, en mi trabajo me he encontrado con escenarios de negocio que requieren lo siguiente: necesito enviar datos a otro sistema en lotes de manera programada diariamente, pero debido a que el sistema está desplegado en clusters, puede causar conflictos de tareas en situaciones uniformes, por lo que es necesario agregar un candado distribuido para garantizar que en un período de tiempo determinado solo un Job pueda completar la tarea programada. En la fase inicial, se consideraron soluciones como la tarea distribuida de ZooKeeper y la programación distribuida de Quartz, pero debido a que ZooKeeper requiere agregar componentes adicionales y Quartz requiere agregar tablas, y ya existe el componente Redis en el proyecto, se considera utilizar el candado distribuido de Redis para completar la función de ocupación de tareas distribuidas.

Registrar el camino torcido que he seguido.

Primera versión:

@Override
	public <T> Long set(String key,T value, Long cacheSeconds) {
		if (value instanceof HashMap) {
			BoundHashOperations valueOperations = redisTemplate.boundHashOps(key);
			valueOperations.putAll((Map) value);
			valueOperations.expire(cacheSeconds, TimeUnit.SECONDS);
		}
		else{
		//Se utiliza un mapa para almacenar
		BoundHashOperations valueOperations = redisTemplate.boundHashOps(key);
		valueOperations.put(key, value);
		//segundos
		valueOperations.expire(cacheSeconds, TimeUnit.SECONDS);
		}
		return null;
	}
	@Override
	public void del(String key) {
		redisTemplate.delete(key);
	}

Se utiliza set y del para realizar la ocupación y liberación del bloqueo, pero después de las pruebas, se descubrió que set no es seguro en términos de hilos, y a menudo lleva a inconsistencias de datos en situaciones de concurrencia.

Segunda versión:

/**
   * Bloqueo distribuido
   * @param range Longitud del bloqueo, permite cuántas solicitudes pueden tomar posesión de los recursos
   * @param key
   * @return
   */
  public boolean getLock(int range, String key) {
    ValueOperations<String, Integer> valueOper1 = template.opsForValue();
    return valueOper1.increment(key, 1) <= range;
  }
  /**
   * Inicializar bloqueo, establecer igual a 0
   * @param key
   * @param expireSeconds
   * @return
   */
  public void initLock(String key, Long expireSeconds) {
    ValueOperations<String, Integer> operations = template.opsForValue();
    template.setKeySerializer(new GenericJackson2JsonRedisSerializer());
    template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
    operations.set(key, 0, expireSeconds * 1000);
  }
  /**
   * Liberar bloqueo
   * @param key
   */
  public void releaseLock(String key) {
    ValueOperations<String, Integer> operations = template.opsForValue();
    template.setKeySerializer(new GenericJackson2JsonRedisSerializer());
    template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
    template.delete(key);
  }

Se realiza la toma de posesión del bloqueo utilizando la operación de incrementación de redis. Pero al liberar el bloqueo, cualquier hilo puede eliminar el valor de la clave en redis. Y el método initLock cubre la operación anterior, por lo que también se desecha este método.

Versión final:

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.jedis.JedisConnection;
import org.springframework.stereotype.Service;
import org.springframework.util.ReflectionUtils;
import redis.clients.jedis.Jedis;
import java.lang.reflect.Field;
import java.util.Collections;
@Service
public class RedisLock {
  private static final String LOCK_SUCCESS = "OK";
  private static final String SET_IF_NOT_EXIST = "NX";
  private static final String SET_WITH_EXPIRE_TIME = "PX";
  private static final Long RELEASE_SUCCESS = 1L;
  @Autowired
  private RedisConnectionFactory connectionFactory;
  /**
   * Intentar obtener el candado distribuido
   * @param lockKey Lock
   * @param requestId Identificador de solicitud
   * @param expireTime Tiempo de expiración
   * @return Si se ha obtenido con éxito
   */
  public boolean lock(String lockKey, String requestId, int expireTime) {
    Field jedisField = ReflectionUtils.findField(JedisConnection.class, "jedis");
    ReflectionUtils.makeAccessible(jedisField);
    Jedis jedis = (Jedis) ReflectionUtils.getField(jedisField, connectionFactory.getConnection());
    String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
    if (LOCK_SUCCESS.equals(result)) {}}
      return true;
    }
    return false;
  }
  /**
   * Liberar candado distribuido
   * @param lockKey Lock
   * @param requestId Identificador de solicitud
   * @return Si se liberó con éxito
   */
  public boolean releaseLock(String lockKey, String requestId) {
    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    Object result = getJedis().eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
    if (RELEASE_SUCCESS.equals(result)) {
      return true;
    }
    return false;
  }
  public Jedis getJedis() {
    Field jedisField = ReflectionUtils.findField(JedisConnection.class, "jedis");
    ReflectionUtils.makeAccessible(jedisField);
    Jedis jedis = (Jedis) ReflectionUtils.getField(jedisField, connectionFactory.getConnection());
    return jedis;
  }
}