Распределенная блокировка Redis

Реализация распределенных блокировок в Redis — как элегантно реализовать распределенные блокировки (JAVA)

Адрес блога https://blog.piaoruiqing.com/2019/05/19/redis распределенная блокировка/

Ключевые слова

• 分布式锁Да:Распределенные системыСинхронизированный общий доступ междуресурспрочь.
• spring-data-redis: Инкапсуляция Spring для Redis, конфигурация проста, и она обеспечивает абстрактную инкапсуляцию для взаимодействия с хранилищем Redis. Это очень элегантно и расширяемо. Рекомендуется прочитать исходный код
• Lua: Lua — это легкий и компактный язык сценариев, который можно выполнять в Redis.

предисловие

В этом документе описывается простая реализация JAVA и оптимизация распределенных блокировок Redis, реализуются такие функции, как автоматическая разблокировка, настраиваемые исключения, повторная попытка и разблокировка заметок, а также делается попытка дополнить распределенные блокировки более элегантным и лаконичным кодом.

необходимость

• Взаимное исключение: в распределенной системной среде блокировка может удерживаться только одним потоком.
• Высокая доступность: блокировка не возникает, и блокировку можно снять со временем, даже если клиент выйдет из строя.
• Неблокирующий: возвращается в случае неудачи в получении блокировки.

план

Redis имеет чрезвычайно высокую производительность, и его команды дружелюбны к распределенной поддержке блокировки.SETКоманда может реализовать обработку блокировки.

SET

• EX seconds -- Set the specified expire time, in seconds.
• PX milliseconds -- Set the specified expire time, in milliseconds.
• NX -- Only set the key if it does not already exist.
• XX -- Only set the key if it already exist.

выполнить

Простая реализация

Практика устанавливается, если не существует (назначение, если она не существует), команда redis является атомарной операцией, поэтому используйте ее отдельноsetНе беспокойтесь о параллелизме, вызывающем исключения при заказе.

Конкретный код реализован следующим образом: (spring-data-redis:2.1.6)

импорт зависимостей

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <version>2.1.4.RELEASE</version>
</dependency>

Настройка перераспределения

@Bean
@ConditionalOnMissingBean
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {

    StringRedisSerializer keySerializer = new StringRedisSerializer();
    RedisSerializer<?> serializer = new StringRedisSerializer();
    StringRedisTemplate template = new StringRedisTemplate();
    template.setConnectionFactory(factory);
    template.setKeySerializer(keySerializer);
    template.setHashKeySerializer(keySerializer);
    template.setValueSerializer(serializer);
    template.setHashValueSerializer(serializer);
    template.afterPropertiesSet();
    return template;
}

Простая реализация распределенной блокировки

/**
 * try lock
 * @author piaoruiqing
 * 
 * @param key       lock key
 * @param value     value
 * @param timeout   timeout
 * @param unit  	time unit
 * @return 
 */
public Boolean tryLock(String key, String value, long timeout, TimeUnit unit) {      

	return redisTemplate.opsForValue().setIfAbsent(key, value, timeout, unit);
}

Приведенный выше код завершает простую функцию распределенной блокировки:

вredisTemplate.opsForValue().setIfAbsent(key, value, timeout, unit);То есть выполнить команду redis:

redis> set dlock:test-try-lock a EX 10 NX
OK
redis> set dlock:test-try-lock a EX 10 NX
null

более ранняя версияspring-data-redisРеализация распределенного блокировки и меры предосторожности

методBoolean setIfAbsent(K key, V value, long timeout, TimeUnit unit);Это было добавлено в версии 2.1.В более ранних версиях setIfAbsent не мог указать время истечения одновременно.Если вы используете его первымsetIfAbsentЕсли вы снова установите срок действия ключа, возникнет риск взаимоблокировки, поэтому в старой версии вам нужно использовать другой метод записи для реализации.spring-data-redis:1.8.20Например

/**
 * try lock
 * @author piaoruiqing
 * 
 * @param key       lock key
 * @param value     value
 * @param timeout   timeout
 * @param unit  	time unit
 * @return 
 */
public Boolean tryLock(String key, String value, long timeout, TimeUnit unit) {

    return redisTemplate.execute(new RedisCallback<Boolean>() {
        @Override
        public Boolean doInRedis(RedisConnection connection) throws DataAccessException {

            JedisCommands commands = (JedisCommands)connection.getNativeConnection();
            String result = commands.set(key, value, "NX", "PX", unit.toMillis(timeout));

            return "OK".equals(result);
        }
    });
}

spring-data-redis:1.8.20Клиент Redis по умолчанию:jedis, доступныйgetNativeConnectionНепосредственно вызовите метод jedis для работы.Конечный эффект новой и старой версий одинаков.

Расширенная оптимизация

Реализация инструмента аннотации распределенной блокировки на основе АОП - не только может, но и проста в использовании

Оптимизация 1 (автоматическая разблокировка и повторение)

Автоматическая разблокировка и повторная попытка: простая реализация распределенных блокировок в предыдущем разделе может удовлетворить основные потребности, но есть еще много улучшений, которые можно оптимизировать.В этом разделе будет оптимизирована автоматическая разблокировка и повторная попытка распределенных блокировок.

Абстрактный класс распределенной блокировки

выполнитьAutoCloseableинтерфейс, доступныйtry-with-resourceЛегко сделать автоматически разблокированным.

/**
 * distributed lock
 * @author piaoruiqing
 *
 * @since JDK 1.8
 */
abstract public class DistributedLock implements AutoCloseable {

    private final Logger LOGGER = LoggerFactory.getLogger(getClass());

    /**
     * release lock
     * @author piaoruiqing
     */
    abstract public void release();

    /*
     * (non-Javadoc)
     * @see java.lang.AutoCloseable#close()
     */
    @Override
    public void close() throws Exception {

        LOGGER.debug("distributed lock close , {}", this.toString());

        this.unlock();
    }
}

Encapsulate redis распределенный замок

RedisDistributedLockЭто абстракция распределенных замков Redis, наследующаяDistributedLockИ реализует интерфейс разблокировки.

/**
 * redis distributed lock
 *
 * @author piaoruiqing
 * @date: 2019/01/12 23:20
 *
 * @since JDK 1.8
 */
public class RedisDistributedLock extends DistributedLock {
    
    private RedisOperations<String, String> operations;
    private String key;
    private String value;
    
    private static final String COMPARE_AND_DELETE =		// (一)
        "if redis.call('get',KEYS[1]) == ARGV[1]\n" +
        "then\n" +
        "    return redis.call('del',KEYS[1])\n" +
        "else\n" +
        "    return 0\n" +
        "end";
    
    /**
     * @param operations
     * @param key
     * @param value
     */
    public RedisDistributedLock(RedisOperations<String, String> operations, String key, String value) {
        this.operations = operations;
        this.key = key;
        this.value = value;
    }
    /*
     * (non-Javadoc)
     * @see com.piaoruiqing.demo.distributed.lock.DistributedLock#release()
     */
    @Override
    public void release() {									// (二)
        List<String> keys = Collections.singletonList(key);
        operations.execute(new DefaultRedisScript<String>(COMPARE_AND_DELETE), keys, value);
    }
	/*
     * (non-Javadoc)
     * @see java.lang.Object#toString()
     */
    @Override
    public String toString() {
        return "RedisDistributedLock [key=" + key + ", value=" + value + "]";
    }
}

• (1): Разблокировать через скрипт Lua, сделать对比锁的值+删除Станьте атомарной операцией, чтобы обеспечить правильность операции разблокировки.Проще говоря, это должно предотвратить删了别人的锁. Например: время блокировки истекло до того, как метод потока А был выполнен, а затем поток В также получил блокировку (с тем же ключом), но в это время, если метод потока А завершает выполнение и пытается разблокироваться, если значение не сравнивается, то A удалит B В это время поток C может снова заблокироваться, и у бизнеса возникнет более серьезная путаница (не слишком полагайтесь на распределенные блокировки. В случае высоких требований к согласованности данных , должна быть выполнена определенная обработка на уровне базы данных, например, уникальные ключи, ограничения, транзакции и т. д. для обеспечения корректности данных)
• (2): использоватьRedisOperationsВыполните сценарий Lua, чтобы разблокировать операцию.
• видетьофициальная документация Redis

реализация метода блокировки

/**
 * @author piaoruiqing
 * @param key           lock key
 * @param timeout       timeout
 * @param retries       number of retries
 * @param waitingTime   retry interval
 * @return
 * @throws InterruptedException
 */
public DistributedLock acquire(String key, long timeout, int retries, long waitingTime) throws InterruptedException {
    final String value 
        = RandomStringUtils.randomAlphanumeric(4) + System.currentTimeMillis(); // (一)
    do {
        Boolean result 
            = stringRedisTemplate.opsForValue().setIfAbsent(key, value, timeout, TimeUnit.MILLISECONDS); // (二)
        if (result) {
            return new RedisDistributedLock(stringRedisTemplate, key, value);
        }
        if (retries > NumberUtils.INTEGER_ZERO) {
            TimeUnit.MILLISECONDS.sleep(waitingTime);
        }
        if(Thread.currentThread().isInterrupted()){
            break;
        }
    } while (retries-- > NumberUtils.INTEGER_ZERO);

    return null;
}

• (1): Значение блокировки должно быть гарантировано уникальным, и использование 4-значной случайной строки + метка времени может в основном соответствовать требованиям. Примечание:UUID.randomUUID()Низкая производительность в ситуациях с высоким параллелизмом.
• (2): Попробуйте заблокировать код версии 2.1, а более ранняя версия относится к реализации из предыдущего раздела.

Этот код уже может удовлетворить потребности автоматической разблокировки и повторной попытки, метод использования:

// 根据key加锁, 超时时间10000ms, 重试2次, 重试间隔500ms
try(DistributedLock lock = redisLockService.acquire(key, 10000, 2, 500);){
    // do something
}

Но он может быть немного более элегантным и инкапсулировать код шаблона, который может поддерживатьLambdaвыражение:

/**
 * lock handler
 * @author piaoruiqing
 *
 * @since JDK 1.8
 */
@FunctionalInterface				// (一)
public interface LockHandler<T> {

    /**
     * the logic you want to execute
     * 
     * @author piaoruiqing
     *
     * @return
     * @throws Throwable
     */
     T handle() throws Throwable;	// (二)
}

• (1): Определить функциональный интерфейс и внедрить бизнес-логикуLambdaВыражения делают код более кратким.
• (2): Исключения в бизнесе не рекомендуются в распределенных блокировках, более разумно бросать напрямую.

использоватьLockHandlerЗавершите реализацию блокировки:

public <T> T tryLock(String key, LockHandler<T> handler, long timeout, boolean autoUnlock, int retries, long waitingTime) throws Throwable {
    try (DistributedLock lock = this.acquire(key, timeout, retries, waitingTime);) {
        if (lock != null) {
            LOGGER.debug("get lock success, key: {}", key);
            return handler.handle();
        }
        LOGGER.debug("get lock fail, key: {}", key);
        return null;
    }
}

На этом этапе кодирование может быть выполнено с использованием распределенных блокировок более элегантным способом:

@Test
public void testTryLock() throws Throwable {
    final String key = "dlock:test-try-lock";
    AnyObject anyObject = redisLockService.tryLock(key, () -> {
        // do something
        return new AnyObject();
    }, 10000, true, 0, 0);
}

Оптимизация 2 (пользовательское исключение)

Пользовательское исключение: инкапсуляция для распределенных блокировок в предыдущей статье может соответствовать большинству бизнес-сценариев, но рассмотрим такую ​​ситуацию, если бизнес сам вернетсяNULLТекущая реализация может иметь неправильную обработку, так как отказ от получения блокировки также вернетNULL, Избегайте возвращенияNULLХотя это решение, оно не подходит для всех сценариев, и в настоящее время это может быть хорошим выбором для поддержки пользовательских исключений.

Его очень легко реализовать, и он добавляется на основе оригинального кода.onFailureПараметры, если блокировка пуста, можно кинуть исключение.

реализация метода блокировки

public <T> T tryLock(String key, LockHandler<T> handler, long timeout, boolean autoUnlock, int retries, long waitingTime, Class<? extends RuntimeException> onFailure) throws Throwable {	// (一)
    try (DistributedLock lock = this.getLock(key, timeout, retries, waitingTime);) {
        if (lock != null) {
            LOGGER.debug("get lock success, key: {}", key);
            return handler.handle();
        }
        LOGGER.debug("get lock fail, key: {}", key);
        if (null != onFailure) {
            throw onFailure.newInstance();	// (二)
        }
        return null;
    }
}

• (один):Class<? extends RuntimeException>ограниченноеonFailureдолжно бытьRuntimeExceptionили его подклассов. Автор считает, что использованиеRuntimeExceptionЕго легче понять семантически.Также при необходимости можно использовать другие исключения (такие как невозможность получения блокировок и необходимость единообразной обработки).
• (2): Отражение

Оптимизация третья (изящно используйте аннотации)

В сочетании с APO для элегантного использования аннотаций для завершения распределенных блокировок:

определить аннотации

Для уменьшения места некоторые заметки свернуты

/**
 * distributed lock
 * @author piaoruiqing
 * @date: 2019/01/12 23:15
 *
 * @since JDK 1.8
 */
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface DistributedLockable {

    /** timeout of the lock */
    long timeout() default 5L;
    
    /** time unit */
    TimeUnit unit() default TimeUnit.MILLISECONDS;
    
    /** number of retries */
    int retries() default 0;

    /** interval of each retry */
    long waitingTime() default 0L;

    /** key prefix */
    String prefix() default "";

    /** parameters that construct a key */
    String[] argNames() default {};

    /** construct a key with parameters */
    boolean argsAssociated() default true;

    /** whether unlock when completed */
    boolean autoUnlock() default true;

    /** throw an runtime exception while fail to get lock */
    Class<? extends RuntimeException> onFailure() default NoException.class;

    /** no exception */
    public static final class NoException extends RuntimeException {

        private static final long serialVersionUID = -7821936618527445658L;

    }
}

• timeout: сверхурочное время
• unit: единица времени
• retries: количество попыток
• waitingTime: время интервала повтора
• prefix: префикс ключа, по умолчанию包名+类名+方法名
• argNames: параметры, составляющие ключ

Аннотации можно использовать в методах.Следует отметить, что аннотации в этой статье реализованы через Spring AOP, поэтому вызовы между методами внутри объекта будут недействительными.

Реализация аспекта

/**
 * distributed lock aspect
 * @author piaoruiqing
 * @date: 2019/02/02 22:35
 *
 * @since JDK 1.8
 */
@Aspect
@Order(10)	// (一)
public class DistributedLockableAspect implements KeyGenerator {	// (二)
    
    private final Logger LOGGER = LoggerFactory.getLogger(getClass());
    @Resource
    private RedisLockClient redisLockClient;
    /**
     * {@link DistributedLockable}
     * @author piaoruiqing
     */
    @Pointcut(value = "execution(* *(..)) && @annotation(com.github.piaoruiqing.dlock.annotation.DistributedLockable)")
    public void distributedLockable() {}
    
    /**
     * @author piaoruiqing
     *
     * @param joinPoint
     * @param lockable
     * @return
     * @throws Throwable
     */
    @Around(value = "distributedLockable() && @annotation(lockable)")
    public Object around(ProceedingJoinPoint joinPoint, DistributedLockable lockable) throws Throwable {
        long start = System.nanoTime();
        final String key = this.generate(joinPoint, lockable.prefix(), lockable.argNames(), lockable.argsAssociated()).toString();
        Object result = redisLockClient.tryLock(
            key, () -> {
                return joinPoint.proceed();
            }, 
            lockable.unit().toMillis(lockable.timeout()), lockable.autoUnlock(), 
            lockable.retries(), lockable.unit().toMillis(lockable.waitingTime()),
            lockable.onFailure()
        );
        long end = System.nanoTime();
        LOGGER.debug("distributed lockable cost: {} ns", end - start);
        return result;
    }
}

• (1): Приоритет соотношения сторон
• (два):KeyGeneratorДля создания стратегии пользовательского ключа используйтеprefix+argName+argВ качестве ключа смотрите конкретную реализациюисходный код.

В настоящее время распределенные блокировки можно использовать с помощью аннотаций, которые содержат меньше кода и являются краткими.

@DistributedLockable(
    argNames = {"anyObject.id", "anyObject.name", "param1"},
    timeout = 20, unit = TimeUnit.SECONDS, 
    onFailure = RuntimeException.class
    )
public Long distributedLockableOnFaiFailure(AnyObject anyObject, String param1, Object param2, Long timeout) {
    try {
        TimeUnit.SECONDS.sleep(timeout);
        LOGGER.info("distributed-lockable: " + System.nanoTime());
    } catch (InterruptedException e) {
    }
    return System.nanoTime();
}

расширять

Существует множество способов реализации распределенных блокировок, и можно выбрать различные носители в соответствии с реальными сценариями и потребностями:

• Redis: высокая производительность, удобная поддержка распределенных блокировок, простота реализации и хорошая производительность в большинстве сценариев.
• Zookeeper: высокая надежность, поддержка распределенных блокировок и удобство, более сложная, но существующая реализация, которую можно использовать.
• База данных: проста в реализации, готова к использованию乐观锁/悲观锁Реализация, средняя производительность, не рекомендуется в сценариях с высокой степенью параллелизма.

Эпилог

В этой статье описывается JAVA-реализация распределенных блокировок Redis и завершаются такие функции, как автоматическая разблокировка, настраиваемые исключения, повторная попытка и разблокировка заметки См. исходный код.адрес.

В этой реализации можно оптимизировать многие вещи, например:

• Реентерабельная блокировка достигается
• Оптимизированная стратегия повторных попыток заключается в подписке на события Redis: подписка на события Redis может дополнительно оптимизировать производительность блокировок.Вы можете использовать wait+notifyAll, чтобы заменить сон в тексте.

Пространство ограничено и будет объяснено позже.

серия статей

1. Начало работы с Redis
2. Redis использует расширенные
3. Распределенная блокировка Redis

использованная литература

• Redis.IO/topics/ День 3…
• Мартин, Читать PP Mann.com/2016/02/08/…