Невозможно сделать всю жизнь удобной и приятной, потому что люди должны обладать способностьюОтношение к невзгодам-- Руссо
Автор Ма Хуа, обычный программист в имперской столице, родом из публичного аккаунта "Source Code Interest Circle"
предисловие
Обычно мы учитываем точки и производительность в собственной реализации распределенной блокировки, вряд ли удастся достичь всеобъемлющего в отрасли распределенной блокировки, чтобы добиться большего успеха, это Redisson.
Однако, если вы вводите необходимость объединения своего собственного проекта, если вы вводите только распределенную блокировку, вы чувствуете, что нет необходимости ее реализовывать; если вы полагаетесь на распределенную функцию в Redisson, то вы можете ссылаться
Прежде чем читать исходный код Redisson, вам необходимо понять происхождение распределенных блокировок, а также преимущества и недостатки настройки каждой реализации.Щелкните ссылку для просмотра
ReentrantLock блокировка повторного входа
Прежде чем мы поговорим о Redisson, давайте поговорим об этомБлокировка повторного входа JDK: ReentrantLock
ReentrantLock обеспечиваетОбщий ресурс JVM позволяет одновременно работать только одним потокам.
Реализовать идеи
Внутренняя справедливая блокировка ReentrantLock и наследование несправедливой блокировкиAQS[AbstractQueuedSynchronizer]
1. Состояние переменной типа int изменено с помощью volatil внутри AQS.Вопросы безопасности потоков и блокировка повторного входа под контролем параллелизма
2. Поместите потоки, которые не конкурируют за блокировки, в очередь AQS для прохожденияLockSupport # парк, разпарковать зависание пробуждение
О просмотре AQS и ROLCадрес ссылки
Redisson
Вы можете напрямую просматривать Githubофициальный сайт РедиссонВведение, для тех из вас, кто не знал этого раньше, вы можете взглянуть на каталог Redisson WIKI и поближе взглянуть на то, как Redis вооружен до зубов Redisson, то есть слишком много каталогов, чтобы поместиться
Вот часть содержания, связанного со статьей
Из предисловия к проекту видно, что уровень людей, написавших это предисловие к проекту, очень вау.Из первого абзаца мы знаем две проблемы.
Что такое Редиссон
Редиссон построен наФреймворк сетки данных в памяти Java на основе Redis, в полной мере воспользоваться рядом преимуществ, предоставляемых базой данных Redis "ключ-значение",На основе общих интерфейсов в наборах утилит JavaПредоставляет пользователюСерия общих инструментов с распределенными функциями
Преимущества Редиссон
Создает оригинальный набор инструментов для координации одномашинных многопоточных параллельных программ.Приобрел способность координировать распределенные многомашинные многопоточные параллельные системы., что значительно снижает сложность проектирования и разработки крупномасштабных распределенных систем.
В то же время, в сочетании с различными распределенными сервисами с богатыми характеристиками, идти дальшеУпрощает взаимодействие между программами в распределенной среде
Это почти то же самое, когда вы это понимаете, поэтому я не буду расширяться.Если вы хотите узнать подробное использование, перейдите в указанный выше каталог.
Блокировка повторного входа Redisson
Поскольку Redisson слишком сложен, большинство разработанных вызовов API связаны с Netty, поэтому здесь толькоКак блокировать, как реализовать повторную блокировку для анализа и как анализировать, когда блокировка продолжается
создать замок
Я здесь, чтобы загрузить исходный код Redisson в местный
Следующая простая программа создает нечестную повторную блокировку с помощью Redisson.
метод lock() успешно заблокированВремя истечения по умолчанию составляет 30 секунд и поддерживает функцию продолжения «сторожевого таймера».
public static void main(String[] args) {
Config config = new Config();
config.useSingleServer()
.setPassword("123456")
.setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);
RLock lock = redisson.getLock("myLock");
try {
lock.lock();
// 业务逻辑
} finally {
lock.unlock();
}
}
Давайте сначала посмотрим на объявление интерфейса RLock
public interface RLock extends Lock, RLockAsync {}
RLock наследует интерфейс Lock в исходном пакете JUC JDK, а также наследует RLockAsync.
RLockAsync буквально означаетПоддержка асинхронного замка, что доказывает, что блокировку можно получить асинхронно
Прочитав исходный код Redisson, вы поймете, что комментарии дороже золота 🙃️
Поскольку существует множество API для получения блокировок, здесь мы используем lock() в качестве объяснения исходного кода, а определение интерфейса довольно простое.
/**
* lock 并没有指定锁过期时间, 默认 30 秒
* 如果获取到锁, 会对锁进行续时
*/
void lock();
получить экземпляр блокировки
Согласно приведенной выше небольшой демонстрации, посмотрите, как выполняется первый шаг получения блокировки.
RLock lock = redisson.getLock("myLock");
// name 就是锁名称
public RLock getLock(String name) {
// 默认创建的同步执行器, (存在异步执行器, 因为锁的获取和释放是有强一致性要求, 默认同步)
return new RedissonLock(connectionManager.getCommandExecutor(), name);
}
Redisson Все команды Redis выполняются ... Исполнителем
После получения привода синхронизации по умолчанию вы инициализируете RedissonLock.
public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
super(commandExecutor, name);
this.commandExecutor = commandExecutor;
// 唯一ID
this.id = commandExecutor.getConnectionManager().getId();
// 等待获取锁时间
this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
// ID + 锁名称
this.entryName = id + ":" + name;
// 发布订阅, 后面关于加、解锁流程会用到
this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
}
попробуй получить замок
Давайте взглянемRLock#lock()Как нижний слой получает блокировку
@Override
public void lock() {
try {
lock(-1, null, false);
} catch (InterruptedException e) {
throw new IllegalStateException();
}
}
leaseTime:время истечения блокировки, -1 использует значение по умолчанию 30 секунд
unit:Единица времени, миллисекунды, секунды, минуты, часы...
interruptibly:Можно ли прервать
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
// 获取当前线程ID
long threadId = Thread.currentThread().getId();
// 🚩 尝试获取锁, 下面重点分析
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// 成功获取锁, 过期时间为空
if (ttl == null) {
return;
}
// 订阅分布式锁, 解锁时进行通知
RFuture<RedissonLockEntry> future = subscribe(threadId);
if (interruptibly) {
commandExecutor.syncSubscriptionInterrupted(future);
} else {
commandExecutor.syncSubscription(future);
}
try {
while (true) {
// 再次尝试获取锁
ttl = tryAcquire(-1, leaseTime, unit, threadId);
// 成功获取锁, 过期时间为空, 成功返回
if (ttl == null) {
break;
}
// 锁过期时间如果大于零, 则进行带过期时间的阻塞获取
if (ttl >= 0) {
try {
// 获取不到锁会在这里进行阻塞, Semaphore, 解锁时释放信号量通知
future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
if (interruptibly) {
throw e;
}
future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
}
// 锁过期时间小于零, 则死等, 区分可中断及不可中断
} else {
if (interruptibly) {
future.getNow().getLatch().acquire();
} else {
future.getNow().getLatch().acquireUninterruptibly();
}
}
}
} finally {
// 取消订阅
unsubscribe(future, threadId);
}
}
Этот фрагмент кода используется для выполнения блокировки, продолжайте просмотр реализации метода.
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
}
Методы lock() и tryLock(...) в конечном итоге вызовут этот метод, который разделен на две ветви процесса.
1. API tryLock(...) асинхронно блокирует и возвращает
2. API lock() и tryLock() асинхронно блокирует и продолжает блокировку
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
// 执行 tryLock(...) 才会进入
if (leaseTime != -1) {
// 进行异步获取锁
return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
// 尝试异步获取锁, 获取锁成功返回空, 否则返回锁剩余过期时间
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
// ttlRemainingFuture 执行完成后触发此操作
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// ttlRemaining == null 代表获取了锁
// 获取到锁后执行续时操作
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
});
return ttlRemainingFuture;
}
Продолжайте смотреть на подробный процесс блокировки tryLockInnerAsync(...), внутренняя форма скрипта Lua обеспечивает атомарные операции.
На данный момент все очень ясно, сценарий Lua упакован Redisoon и, наконец, передан через Netty.
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
return evalWriteAsync(getName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);",
Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}
evalWriteAsync(...) — это инкапсуляция команды Eval, и приложение Netty не будет продолжать ее выполнение.
Заблокировать Луа
Выполните сценарий Lua, заблокированный Redis, и сделайте снимок экрана, чтобы все могли увидеть параметры и их конкретное значение.
KEYS[1]: myLock
ARGV[1]: 36000... Это проверенное мной время истечения в миллисекундах
ARGV[2]: UUID + идентификатор потока
# KEYS[1] 代表上面的 myLock
# 判断 KEYS[1] 是否存在, 存在返回 1, 不存在返回 0
if (redis.call('exists', KEYS[1]) == 0) then
# 当 KEYS[1] == 0 时代表当前没有锁
# 使用 hincrby 命令发现 KEYS[1] 不存在并新建一个 hash
# ARGV[2] 就作为 hash 的第一个key, val 为 1
# 相当于执行了 hincrby myLock 91089b45... 1
redis.call('hincrby', KEYS[1], ARGV[2], 1);
# 设置 KEYS[1] 过期时间, 单位毫秒
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
# 查找 KEYS[1] 中 key ARGV[2] 是否存在, 存在回返回 1
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
# 同上, ARGV[2] 为 key 的 val +1
redis.call('hincrby', KEYS[1], ARGV[2], 1);
# 同上
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
# 返回 KEYS[1] 过期时间, 单位毫秒
return redis.call('pttl', KEYS[1]);
Блок-схема всей блокировки Lua-скрипта выглядит следующим образом:
Теперь оглянитесь назад и посмотрите, как блокировка откладывается после получения блокировки.
Когда заблокировано
Там в принципе тоже самое и друзья говорили на эту тему, идеи и воплощенные до Редиссона
Поговорим о конкретных идеях реализации Redisson, в китайском переводе это называется «сторожевой пес».
1,Выполнить процесс «сторожевой таймер» после получения блокировки
2,Используйте тайм-аут Netty для реализации временной задержки
3.Например, если блокировка истекает в течение 30 секунд, то каждые 1/3 времени, то есть 10 секунд, будет проверяться, существует ли блокировка, и если она существует, обновлять время ожидания блокировки.
Некоторые друзья могут задать такой вопрос,Что, если проверка существует, а блокировка снимается только после истечения срока действия установленной блокировки?
Если есть такой вопрос, представитель действительно рассмотрел все возможные ситуации, но не беспокойтесь об этом.
Проверка и установка операций времени истечения срока действия, выполняемых сценариями Lua, используемыми в Redisson., сам по себе является атомарным и вышеуказанная ситуация не возникает
Если вы не хотите ссылаться на пакеты Netty, такие как пакеты, использующие инструменты очереди задержки, также можно сделать «сторожевой таймер».
Вот также связанный код, который может позволить друзьям более интуитивно понять, как зафиксировать время.
Я такой горячий парень, набери кодRedissonLock#tryAcquireAsync(...)
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
// ...
// 尝试异步获取锁, 获取锁成功返回空, 否则返回锁剩余过期时间
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
// ttlRemainingFuture 执行完成后触发此操作
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// 获取到锁后执行续时操作
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
});
return ttlRemainingFuture;
}
Вы можете видеть, что метод продолжения использует threadId в качестве идентификатора для продолжения.
private void scheduleExpirationRenewal(long threadId) {
ExpirationEntry entry = new ExpirationEntry();
ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
if (oldEntry != null) {
oldEntry.addThreadId(threadId);
} else {
entry.addThreadId(threadId);
renewExpiration();
}
}
Хорошо знать основную концепцию, нет необходимости изучать каждую строку кода.
private void renewExpiration() {
ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ee == null) {
return;
}
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ent == null) {
return;
}
Long threadId = ent.getFirstThreadId();
if (threadId == null) {
return;
}
RFuture<Boolean> future = renewExpirationAsync(threadId);
future.onComplete((res, e) -> {
if (e != null) {
log.error("Can't update lock " + getName() + " expiration", e);
return;
}
if (res) {
// 调用本身
renewExpiration();
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
ee.setTimeout(task);
}
Разблокировать операцию
Операция разблокировки относительно проста или относительно проста.
@Override
public void unlock() {
try {
get(unlockAsync(Thread.currentThread().getId()));
} catch (RedisException e) {
if (e.getCause() instanceof IllegalMonitorStateException) {
throw (IllegalMonitorStateException) e.getCause();
} else {
throw e;
}
}
}
После успешной разблокировки предыдущий тайм-аут «сторожевого таймера» будет отменен и возвращен к успеху.
@Override
public RFuture<Void> unlockAsync(long threadId) {
RPromise<Void> result = new RedissonPromise<Void>();
RFuture<Boolean> future = unlockInnerAsync(threadId);
future.onComplete((opStatus, e) -> {
// 取消自动续时功能
cancelExpirationRenewal(threadId);
if (e != null) {
// 失败
result.tryFailure(e);
return;
}
if (opStatus == null) {
IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
+ id + " thread-id: " + threadId);
result.tryFailure(cause);
return;
}
// 解锁成功
result.trySuccess(null);
});
return result;
}
Еще один существенный момент, определение разблокированного сценария Lua.
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
"else " +
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end; " +
"return nil;",
Arrays.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}
Или подойди к картинке, чтобы понять, скрипт Lua будет подробно разобран
Разблокировать Луа
Старые правила, картинки и описания параметров
KEYS[1]: myLock
KEYS[2]: redisson_lock_channel:{myLock}
ARGV[1]: 0
ARGV[2]: 360000... (Срок действия)
ARGV [3]: 7f0c54e2 ... (блокировка хэша ключа)
# 判断 KEYS[1] 中是否存在 ARGV[3]
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
return nil;
end;
# 将 KEYS[1] 中 ARGV[3] Val - 1
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
# 如果返回大于0 证明是一把重入锁
if (counter > 0) then
# 重制过期时间
redis.call('pexpire', KEYS[1], ARGV[2]);
return 0;
else
# 删除 KEYS[1]
redis.call('del', KEYS[1]);
# 通知阻塞等待线程或进程资源可用
redis.call('publish', KEYS[2], ARGV[1]);
return 1;
end;
return nil;
Алгоритм Редлока
Нельзя отрицать, что распределенная блокировка, разработанная Redisson, действительно NB, но она до сих пор не решена.Проблема потери блокировки, вызванная асинхронной синхронизацией данных на узле master-slave
Итак, автор Redis Antirez запустилАлгоритм красного замка, суть этого алгоритма такова:Нет подчиненного узла.Если развернуто несколько Redis, экземпляры независимы друг от друга, и нет репликации ведущий-подчиненный или других механизмов координации кластера.
как пользоваться
Создайте несколько узлов Redisson и сформируйте полную распределенную блокировку из этих несвязанных узлов.
public static void main(String[] args) {
String lockKey = "myLock";
Config config = new Config();
config.useSingleServer().setPassword("123456").setAddress("redis://127.0.0.1:6379");
Config config2 = new Config();
config.useSingleServer().setPassword("123456").setAddress("redis://127.0.0.1:6380");
Config config3 = new Config();
config.useSingleServer().setPassword("123456").setAddress("redis://127.0.0.1:6381");
RLock lock = Redisson.create(config).getLock(lockKey);
RLock lock2 = Redisson.create(config2).getLock(lockKey);
RLock lock3 = Redisson.create(config3).getLock(lockKey);
RedissonRedLock redLock = new RedissonRedLock(lock, lock2, lock3);
try {
redLock.lock();
} finally {
redLock.unlock();
}
}
Конечно, в алгоритме Redlock нет никаких сомнений.Вы можете зайти на официальный сайт Redis, чтобы увидеть дебаты между Мартином Клеппманном и автором Redis Антирезом.
Компромиссы между принципами CAP
Принцип CAP, также известный как теорема CAP, относится в распределенной системе к Consistency (непротиворечивости), Availability (доступности), Partition trust (допуск к разделению),Три не могут быть объединены
Консистенция (С):Имеют ли все резервные копии данных в распределенной системе одно и то же значение в одно и то же время (эквивалентно тому, что все узлы обращаются к одной и той же самой последней копии данных).
Доступность (А):После сбоя некоторых узлов в кластере может ли кластер в целом отвечать на запросы клиента на чтение и запись (с высокой доступностью для обновлений данных).
Допуск перегородки (P):На практике разделение эквивалентно ограничению времени для связи.Если система не может достичь согласованности данных в течение срока, это означает, что произошло разделение, и необходимо сделать выбор между C и A для текущей операции.
Распределенная блокировка
Если вы хотите обеспечить строгую согласованность между вышеупомянутыми распределенными блокировками, вы можете использовать распределенные блокировки Zookeeper,Благодаря базовому протоколу ZAB (протокол атомарной широковещательной рассылки) он, естественно, удовлетворяет требованиям CP.
Но это также означает падение производительности, поэтому просмотр Redis и Zookeeper без просмотра конкретных данных представляет собой компромисс между производительностью и согласованностью.
Если проект не сильно зависит от ZK, хорошо использовать Redis, т.к. Redis сейчас широко используется, а на Redis ссылаются в большинстве проектов
Для этого нет необходимости вводить новый компонент.Если бизнес-сценарий невыносим для потери блокировки, вызванной синхронными данными асинхронного режима Redis, можно справиться с этим на бизнес-уровне.
последние слова
Недавно я пишу о мульти-резьбовом исходном коде, а анализ исходного кода под JUC будет выводиться позже.
1. Защелка обратного отсчета
2. Локальный поток
3. Связанные с атомом
Включая две последние статьи о распределенных блокировках, их длина относительно велика, я надеюсь, что вы сможете терпеливо просмотреть их.
Надеюсь на обратную связь и исправление ошибок в статье🙏, любовь моих друзей - самая большая поддержка для меня, и, наконец, надеюсь, что вы сможетеСтавьте лайки, комментируйте, смотрите Санлиан!
Рекомендуемое чтение:
- Краткая статья | Как решить задачу быстрого потребления, когда пул потоков JDK не превышает максимальное количество потоков
- 4D Графика | Поговорим о ReentrantLock и AQS (после прочтения вы меня не найдете)
- Как обрабатывать исключения выполнения потоков в пуле потоков JDK?
- Используйте новую функцию Java8 parallelStream с осторожностью