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

Архитектура

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

Я давно говорил о блокировках в параллельном программировании.Механизм блокировки для параллельного программирования: синхронизированный и блокирующий. В системе с одним процессом, когда есть несколько потоков, которые могут изменять переменную одновременно, необходимо синхронизировать переменную или блок кода, чтобы он мог выполняться линейно при изменении такой переменной, чтобы исключить одновременное изменение переменной. . Суть синхронизации достигается за счет блокировок. Чтобы понять, что несколько потоков могут выполнять один и тот же блок кода только одним потоком за раз, необходимо где-то сделать метку.Эта метка должна быть видна каждому потоку.Если метка не существует, ее можно установить , а другие последующие потоки обнаруживают, что метка уже есть, и ждут, пока поток с меткой завершит блок кода синхронизации, чтобы отменить метку, а затем попытаются установить метку.

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

Распространенным сценарием является сценарий seckill, в котором служба заказов развертывает несколько экземпляров. Например, есть 4 продукта seckill, первый пользователь покупает 3, а второй пользователь покупает 2. В идеале первый пользователь может купить успешно, а второй пользователь подсказывает, что покупка не удалась, и наоборот. Фактическая ситуация, которая может возникнуть, заключается в том, что оба пользователя получают инвентарь 4, а первый пользователь покупает 3. Перед обновлением инвентаря второй пользователь разместил заказ на 2 товара, а обновленный инвентарь был 2, что привело к ошибке.

В приведенном выше сценарии запасы товаров являются общей переменной.В условиях высокой параллелизма необходимо обеспечить, чтобы доступ к ресурсам был взаимоисключающим. В автономной среде Java фактически предоставляет множество API, связанных с параллельной обработкой, но эти API бессильны в распределенных сценариях. То есть чистый Java Api не может обеспечить возможность распределенной блокировки. В распределенной системе из-за распределенной природы распределенной системы, то есть многопоточности и многопроцессорности и распределения на разных машинах, две блокировки синхронизированной и блокировки потеряют эффект исходной блокировки, и нам нужно реализовать распределенные блокировки самостоятельно.

Общие схемы запирания следующие:

  • Распределенная блокировка на основе базы данных
  • На основе кеша внедрите распределенные блокировки, такие как Redis
  • Распределенная блокировка на основе Zookeeper

Ниже мы кратко представим реализацию этих типов замков.

на основе базы данных

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

Добавления и удаления на основе таблиц базы данных

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

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

После завершения выполнения необходимо удалить запись.

Конечно, это всего лишь краткое введение здесь. Вышеупомянутая схема может быть оптимизирована, например, путем применения базы данных master-slave, двусторонней синхронизации между данными. Как только он зависнет, быстро переключитесь на резервную базу данных, выполните запланированную задачу и очистите данные тайм-аута в базе данных через регулярные промежутки времени, используйте цикл while до тех пор, пока вставка не будет успешной, а затем успешно выполните возврат, хотя это не рекомендуется; вы также можете записать текущую информацию о хосте и информацию о потоке машины, получившей блокировку, а затем сначала запросить базу данных при получении блокировки в следующий раз.Если информацию о хосте и информацию о потоке текущей машины можно найти в базе данных, просто назначьте блокировку ему напрямую.

Эксклюзивная блокировка базы данных

Мы также можем реализовать распределенные блокировки с помощью монопольных блокировок базы данных. Основываясь на движке MySql InnoDB, вы можете использовать следующие методы для реализации операций блокировки:

public void lock(){
    connection.setAutoCommit(false)
    int count = 0;
    while(count < 4){
        try{
            select * from lock where lock_name=xxx for update;
            if(结果不为空){
                //代表获取到锁
                return;
            }
        }catch(Exception e){
        }
        //为空或者抛异常的话都表示没有获取到锁
        sleep(1000);
        count++;
    }
    throw new LockException();
}

Добавьте для обновления после оператора запроса, и база данных добавит монопольную блокировку к таблице базы данных во время процесса запроса. Когда монопольная блокировка добавляется к записи, другие потоки не могут добавить монопольную блокировку к строке. Другие, которые не получат блокировку, будут заблокированы в приведенном выше операторе select.Есть два возможных результата: блокировка получена до тайм-аута и блокировка не получена до тайм-аута.

Поток, который получает эксклюзивную блокировку, может получить распределенную блокировку.Когда блокировка получена, бизнес-логика метода может быть выполнена.После выполнения метода блокировка снимается.connection.commit().

Существующие проблемы в основном связаны с низкой производительностью и ненормальным временем ожидания sql.

Преимущества и недостатки блокировки базы данных

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

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

На основе зоопарка

Распределенные блокировки, которые могут быть реализованы временными упорядоченными узлами на базе zookeeper. Когда каждый клиент блокирует метод, в каталоге указанного узла создается уникальный мгновенный упорядоченный узел, соответствующий методу на zookeeper. Способ определить, следует ли приобретать блокировку, очень прост, вам нужно только определить блокировку с наименьшим порядковым номером в заказанном узле. При снятии блокировки просто удалите переходный узел. В то же время это может избежать проблемы взаимоблокировки, когда блокировка не может быть снята из-за простоя службы.

Предоставляемые сторонние библиотекиcurator, в конкретном использовании читатель может убедиться сам. InterProcessMutex, предоставляемый Curator, представляет собой реализацию распределенных блокировок. Метод Acquire получает блокировку, а метод Release снимает блокировку. Кроме того, могут быть эффективно решены такие проблемы, как снятие блокировки, блокировка блокировки и блокировка с повторным входом. Поговорим о реализации блокирующих блокировок.Клиент может создавать последовательные узлы в ZK и привязывать слушателей к узлам.Как только узлы изменятся, Zookeeper уведомит клиента, и клиент сможет проверить, принадлежат ли в данный момент созданные им узлы. с наименьшим серийным номером может выполнять бизнес-логику, если получает блокировку.

Наконец, распределенная блокировка, реализованная Zookeeper, на самом деле имеет недостаток, то есть производительность может быть не такой высокой, как у службы кэширования. Потому что каждый раз в процессе создания и снятия блокировки мгновенные узлы должны динамически создаваться и уничтожаться для реализации функции блокировки. Создание и удаление нод в ZK возможно только через сервер-лидер, и тогда данные не могут быть разделены со всеми машинами-последователями. Из-за проблем с параллелизмом может быть сетевой джиттер, и сеансовое соединение между клиентом и кластером ZK разрывается.Кластер ZK считает, что клиент зависает и удаляет временный узел.В это время другие клиенты могут получить распределенные блокировки.

на основе кэша

По сравнению со схемой реализации распределенной блокировки на основе базы данных, реализация на основе кеша будет лучше с точки зрения производительности, а скорость доступа будет намного выше. И многие кэши могут быть развернуты в кластерах, что может решить одноточечные проблемы. Существует несколько типов блокировок на основе кеша, таких как memcached и redis.В этой статье в основном объясняется распределенная реализация на основе redis.

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

SETNX

Используя SETNX Redis для реализации распределенных блокировок, несколько процессов выполняют следующие команды Redis:

SETNX lock.id <current Unix time + lock timeout + 1>

SETNX устанавливает значение ключа в значение тогда и только тогда, когда ключ не существует. Если данный ключ уже существует, SETNX ничего не делает.

  • Возвращает 1, указывая на то, что процесс получил блокировку, и SETNX устанавливает значение ключа lock.id равным времени тайм-аута блокировки, текущему времени + времени действия блокировки.
  • Возвращает 0, указывая на то, что другие процессы получили блокировку и процесс не может войти в критическую секцию. Процесс может продолжать выполнять операции SETNX в цикле, чтобы получить блокировку.

тупиковая проблема

SETNX реализует распределенные блокировки, поэтому возможны взаимоблокировки. По сравнению с блокировкой в ​​автономном режиме, в распределенной среде необходимо не только обеспечить видимость процесса, но и учитывать сетевую проблему между процессом и блокировкой. После того, как поток получает блокировку, он отключается от Redis, блокировка не снимается вовремя, и другие потоки, конкурирующие за блокировку, зависают, что приводит к взаимоблокировке.

При использовании SETNX для получения блокировки мы устанавливаем значение ключа lock.id равным действительному времени блокировки. После того, как поток получит блокировку, другие потоки продолжат определять, истекло ли время блокировки. Если время ожидания истекло , ожидающий поток также будет иметь возможность получить блокировку. Однако время блокировки истекает, и мы не можем просто использовать команду DEL, чтобы удалить ключ lock.id, чтобы снять блокировку.

Рассмотрим следующие сценарии:

  1. A сначала получает блокировку lock.id, а затем линия A отключается. Оба B и C ждут, чтобы конкурировать за замок;
  2. B, C считывают значение lock.id, сравнивают текущее время со значением ключа lock.id, чтобы определить, истекло ли время ожидания, и обнаруживают, что оно истекло;
  3. B выполняет команду DEL lock.id и выполняет команду SETNX lock.id и возвращает 1, B получает блокировку;
  4. Поскольку C только что обнаружил, что время ожидания блокировки истекло, он выполняет команду DEL lock.id, удаляет ключ lock.id, только что установленный B, выполняет команду SETNX lock.id и возвращает 1, то есть C получает ключ замок.

Вышеупомянутые шаги, очевидно, имеют проблему, из-за которой B и C получают блокировку одновременно. После обнаружения тайм-аута блокировки поток не может просто удалить ключ с помощью DEL, чтобы получить блокировку.

Для улучшения вышеперечисленных шагов проблема заключается в операции удаления ключа, так как же улучшить ее после получения блокировки?
Сначала посмотрите на операцию GETSET в Redis,GETSET key value, устанавливает значение данного ключа в value и возвращает старое значение ключа. Используя эту команду операции, мы улучшаем вышеуказанные шаги.

  1. A сначала получает блокировку lock.id, а затем линия A отключается. Оба B и C ждут, чтобы конкурировать за замок;
  2. B, C считывают значение lock.id, сравнивают текущее время со значением ключа lock.id, чтобы определить, истекло ли время ожидания, и обнаруживают, что оно истекло;
  3. B обнаруживает, что время блокировки истекло, то есть текущее время больше, чем значение ключа lock.id, B выполнит
    GETSET lock.id <current Unix timestamp + lock timeout + 1>Установите метку времени и оцените, получил ли процесс блокировку, сравнив, меньше ли старое значение ключа lock.id, чем текущее время;
  4. B обнаруживает, что значение, возвращаемое GETSET, меньше текущего времени, выполняет команду DEL lock.id и команду SETNX lock.id и возвращает 1, а B получает блокировку;
  5. C выполняет GETSET, и полученное время превышает текущее время, затем продолжайте ждать.

Прежде чем поток снимет блокировку, то есть выполнит операцию DEL lock.id, он должен определить, истекло ли время блокировки. Если время блокировки истекло, блокировка может быть получена другим потоком, и выполнение операции DEL lock.id напрямую приведет к снятию блокировки, полученной другими потоками.

реализация

получить замок

public boolean lock(long acquireTimeout, TimeUnit timeUnit) throws InterruptedException {
    acquireTimeout = timeUnit.toMillis(acquireTimeout);
    long acquireTime = acquireTimeout + System.currentTimeMillis();
    //使用J.U.C的ReentrantLock
    threadLock.tryLock(acquireTimeout, timeUnit);
    try {
    	//循环尝试
        while (true) {
        	//调用tryLock
            boolean hasLock = tryLock();
            if (hasLock) {
                //获取锁成功
                return true;
            } else if (acquireTime < System.currentTimeMillis()) {
                break;
            }
            Thread.sleep(sleepTime);
        }
    } finally {
        if (threadLock.isHeldByCurrentThread()) {
            threadLock.unlock();
        }
    }
    return false;
}
public boolean tryLock() {
    long currentTime = System.currentTimeMillis();
    String expires = String.valueOf(timeout + currentTime);
    //设置互斥量
    if (redisHelper.setNx(mutex, expires) > 0) {
    	//获取锁，设置超时时间
        setLockStatus(expires);
        return true;
    } else {
        String currentLockTime = redisUtil.get(mutex);
        //检查锁是否超时
        if (Objects.nonNull(currentLockTime) && Long.parseLong(currentLockTime) < currentTime) {
            //获取旧的锁时间并设置互斥量
            String oldLockTime = redisHelper.getSet(mutex, expires);
            //旧值与当前时间比较
            if (Objects.nonNull(oldLockTime) && Objects.equals(oldLockTime, currentLockTime)) {
            	//获取锁，设置超时时间
                setLockStatus(expires);
                return true;
            }
        }
        return false;
    }
}

lock вызывает метод tryLock с параметрами полученного таймаута и единицы В течение периода таймаута поток получает блокировку и крутится там до тех пор, пока держатель спин-блокировки не освободит блокировку.

В методе tryLock основная логика следующая:

  • setnx(lockkey, текущее время + истечение срока действия), если он возвращает 1, блокировка получена успешно; если он возвращает 0, блокировка не получена
  • get(lockkey) получает значение oldExpireTime и сравнивает это значение с текущим системным временем, если оно меньше текущего системного времени, считается, что время блокировки истекло и другие запросы могут быть получены повторно
  • Вычислить newExpireTime = текущее время + время истечения срока действия, затем getset(lockkey, newExpireTime) вернет значение текущего ключа блокировки currentExpireTime
  • Определите, равны ли currentExpireTime и oldExpireTime. Если они равны, текущая настройка getset выполнена успешно и блокировка получена. Если он не равен, это означает, что блокировка была получена другим запросом, тогда текущий запрос может сразу вернуться к ошибке или продолжить повторную попытку.

разблокировать замок

public boolean unlock() {
    //只有锁的持有线程才能解锁
    if (lockHolder == Thread.currentThread()) {
        //判断锁是否超时，没有超时才将互斥量删除
        if (lockExpiresTime > System.currentTimeMillis()) {
            redisHelper.del(mutex);
            logger.info("删除互斥量[{}]", mutex);
        }
        lockHolder = null;
        logger.info("释放[{}]锁成功", mutex);
        return true;
    } else {
        throw new IllegalMonitorStateException("没有获取到锁的线程无法执行解锁操作");
    }
}

Фактически, при реализации описанного выше получения блокировки функция снятия блокировки здесь не нужна.Заинтересованные читатели могут скомбинировать приведенный выше код, чтобы понять, почему? Вы можете оставить сообщение, если у вас есть идея!

Суммировать

В этой статье в основном объясняется реализация распределенных блокировок на основе Redis.В распределенной среде проблема согласованности данных всегда была относительно важной темой, а синхронизированные и блокирующие блокировки потеряли свой эффект в распределенной среде. Общие схемы блокировки включают распределенные блокировки на основе базы данных, распределенные блокировки на основе кеша и распределенные блокировки на основе Zookeeper.Кратко представлены характеристики реализации каждой блокировки, затем в документе исследуется схема реализации блокировок Redis, и, наконец, в этой статье распределенная блокировка Redis, основанная на реализации Java, читатели могут убедиться в этом сами.

Ссылаться на

  1. Немного понимания распределенных блокировок
  2. Распределенный замок 1 Общие технические решения Java
  3. Несколько реализаций распределенных блокировок
Категории