A, simple use
1.1 Importing Dependencies
<dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.13.5</version>Copy the code1.2 Basic Code
public static void main(String[] args) throws InterruptedException { // 1. Create config object Config config = new Config(); Config.useclusterservers () // Use cluster model.setscanInterval (2000) // Cluster scan status in milliseconds. AddNodeAddress ("redis://127.0.0.1:7181"); // 2. Create Redisson instance RedissonClient redisson = Redisson.create(config); RLock lock = redisson.getLock("anyLock"); Lock. Lock (); lock.unlock(); Lock (10, timeunit.seconds); Boolean res = lock.tryLock(100, 10, timeunit.seconds); if (res) { try { // do } finally { lock.unlock(); }}}Copy the codeSee Wiki for details
Second, source code analysis
2.1 Reentrant lock
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]);
Copy the codeLua script: Lua script
First check whether KEYS[1] exist, and if so, set a hash data structure, then set an expiration date return.
If KEYS[1] already exist, add 1 to the key, reset the expiration time, calculate how long will expire, and return.
KEYS[1] is the path anyLock we set in the code. With Redisson, each client will have its own Manager class, which will have its own UUID, ARGV[2], The current UUID:threadId, the current client id, the current threadId on the concatenation.
Because redisson has a watchdog mechanism, the default is 30000ms, this implementation of the function is that if you have a lock on the client, and the watchdog check, and the client is still alive, The watchdog then performs the renewal operation, which means that ARGV[1] is 3000ms.
The Redisson mode used here is cluster mode. Therefore, during the execution of Lua script, the algorithm will first calculate which slot we need to be in and which node the slot belongs to. Finally, the command will be executed on this node.
2.2 the watchdog
This mechanism guarantees that the lock will be renewed for 30 seconds at a time if the lock client still exists. If the machine holding the lock is down, then the watchdog on the machine will not execute, do not execute, the lock time will slowly expire, release the lock, or at most 30 seconds.
Basically, once the client has acquired the lock, it will trigger a scheduling task that will be called every 10 seconds
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) {
// reschedule itself
renewExpiration();
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
Copy the codeThe main core is the following Lua script
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('pexpire', KEYS[1], ARGV[1]); return 1; end; return 0; .Copy the codeKEYS[1] is the path we set, ARGV[2] is the concatenation of the current client UUID and thread ID. The default expiration time is set to 30 seconds.
When the client releases the lock, the key is no longer there, and the task stops, as well as if the client is down, so that the key is not renewed and the rest of the client waits up to 30 seconds before trying to acquire the lock.
2.3 Synchronous Blocking
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]);
Copy the codeIf a client has already acquired the lock and another client has acquired the lock, the key will be the id of the current client plus the ID of the current thread. In this case, both ifs are not satisfied, and the remaining time of the current primary key will be returned.
If the lock was successfully acquired, it will return nil, and if it returns anything else, it will fail to lock, so it will return the remaining time, which is not nil, so it will go through this while(true) loop, which is basically trying to get the lock again, and if it doesn’t get the lock again, It will wait for the TTL time to acquire the lock, blocking until the lock is acquired.
while (true) { ttl = tryAcquire(-1, leaseTime, unit, threadId); // lock acquired if (ttl == null) { break; } // waiting for message if (ttl >= 0) { try { 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(); }}}Copy the code2.4 release the lock
Locks are automatically released during downtime
If the machine goes down, then the watchdog’s timing task is gone, that is, after 30 seconds at most, the key expires and the lock is automatically released.
Active release lock
"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));Copy the codeThis lua script is the key to unlocking the lock,
-
-
- KEYS[1] = the key we originally set,
- KEYS[2] = getChannelName() = redisson_lock__channel_KEYS[1]
- ARGV[1] = LockPubSub.UNLOCK_MESSAGE = 0L
- ARGV[2] = internalLockLeaseTime = 30S
- ARGV[3] = getLockName(threadId) = UUID of current lock client concatenates current threadId
-
If the lock exists, return null; if the lock does not exist, return null.
Then, if the key exists, it will decrement the value of the lock and determine the result. If it is equal to 0, it will only lock once. In this case, it will delete the key directly with the del command and publish a message with the publish subscribe command.
If it is not equal to zero, it means that the reentrant lock is added multiple times, and then an expiration time is refreshed to 30 seconds.
2.5 Attempts to obtain timeout and automatic release of timeout lock
Redisson provides a higher-order usage, that is, try 100s and give up if 100s can’t get the lock; If a lock is acquired, the lock is held for 10 seconds. If the lock is not released for 10 seconds, the lock is automatically released.
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
Copy the codelong time = unit.toMillis(waitTime); Long current = system.currentTimemillis (); long threadId = Thread.currentThread().getId(); Long TTL = tryAcquire(waitTime, leaseTime, unit, threadId); // lock acquired if (ttl == null) { return true; } time -= system.currentTimemillis () -current; if (time <= 0) { acquireFailed(waitTime, unit, threadId); return false; } current = System.currentTimeMillis(); . . . Time -= system.currentTimemillis () -current; if (time <= 0) { acquireFailed(waitTime, unit, threadId); return false; } currentTime = system.currentTimemillis (); currentTimeMillis(); ttl = tryAcquire(waitTime, leaseTime, unit, threadId); // lock acquired if (ttl == null) { return true; } time -= System.currentTimeMillis() - currentTime; if (time <= 0) { acquireFailed(waitTime, unit, threadId); return false; } // waiting for message currentTime = System.currentTimeMillis(); if (ttl >= 0 && ttl < time) { subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } else { subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS); } time -= System.currentTimeMillis() - currentTime; if (time <= 0) { acquireFailed(waitTime, unit, threadId); return false; } } } finally { unsubscribe(subscribeFuture, threadId); }Copy the codeTry to get lock timeout:
If it fails to acquire the lock, enter a while(true) loop and try to acquire the lock again. Its logic is the same as the above logic. If it fails to acquire the lock, the remaining time will be updated. Block until the remaining time returns to zero, and false is returned indicating that the lock failed to be acquired.
Automatic release of timeout:
This logic is associated with the attempt to acquire the lock logic, when attempting to acquire the lock, there is this code
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);Copy the codeIf we are not using this high order usage and the value of leaseTime is -1, then we will not walk in the if and we will go to the following one, and leaseTime will become the default value of 30000ms. But in this case, we set the lock expiration time to be 10s. Lease = 10000ms. When a lock is added, the key of the lock is set and its expiration time is set to 10s.
If we set leaseTime, we will return it directly. If we use the default value, we will judge whether we need to add the task according to the return value of the lock, that is watchdog. It will not join the watchdog, it only has a survival time of 10s, to the time, will be released automatically.
2.6 hidden trouble
If the client has just written a lock on the master, the master breaks down, and the master asynchronously synchronizes the lock to the slave, the slave becomes the master. If another client is locking, the slave will successfully acquire the lock. The problem of two clients holding the same distributed lock can cause some data problems.
2.7 Reentrant lock summary
- Lock: Set the hash structure in Redis. The default expiration time is 30,000 ms.
- Maintain lock: A background scheduling task is scheduled every 10 seconds. As long as both the client and the key exist, the current key expiration time will be updated.
- Lock mutexes: Another client or thread trying to lock will be stuck in a while(true) loop, waiting.
- Reentrant lock: The same thread can be locked more than once, increasing the hash structure by 1 each time.
- Manually release the lock: Decrease the value by 1 on the hash structure and check whether the remaining numbers are 0. If the remaining numbers are 0, the key is directly deleted.
- Release lock: When the client is down, background scheduling tasks are cancelled and the key expiration time is not refreshed. The key automatically disappears after 30 seconds by default.
- Lock timeout attempt: in a loop, attempts to acquire the lock are kept. If the lock is not acquired, the loop exits, returning false.
- Automatic lock release: Set the timeout period when locking, so that no scheduled tasks will be performed, and the key will expire after the set expiration time.
Third, fair lock
3.1 lock
RLock fairLock = redisson.getFairLock("anyLock"); // Fairlock. lock();Copy the codeThis lua script implements the basic logic using the set and list data structures
"while true do " + "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" + "if firstThreadId2 == false then " + "break;" + "end; " + "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" + "if timeout <= tonumber(ARGV[4]) then " + "redis.call('zrem', KEYS[3], firstThreadId2); " + "redis.call('lpop', KEYS[2]); " + "else " + "break;" + "end; " + "end;" + "if (redis.call('exists', KEYS[1]) == 0) and ((redis.call('exists', KEYS[2]) == 0) " + "or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then " + "redis.call('lpop', KEYS[2]); " + "redis.call('zrem', KEYS[3], ARGV[2]); " + "redis.call('hset', 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; " + "local firstThreadId = redis.call('lindex', KEYS[2], 0); " + "local ttl; " + "if firstThreadId ~= false and firstThreadId ~= ARGV[2] then " + "ttl = tonumber(redis.call('zscore', KEYS[3], firstThreadId)) - tonumber(ARGV[4]);" + "else " + "ttl = redis.call('pttl', KEYS[1]);" + "end; " + "local timeout = ttl + tonumber(ARGV[3]);" + "if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then " + "redis.call('rpush', KEYS[2], ARGV[2]);" + "end; " + "return ttl;" , Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName), internalLockLeaseTime, getLockName(threadId), currentTime + threadWaitTime, currentTime);Copy the code| KEYS[1] | getName() | Set lock key (anyLock) |
|---|---|---|
| KEYS[2] | threadsQueueName | redisson_lock_queue:{anyLock} |
| KEYS[3] | timeoutSetName | redisson_lock_timeout:{anyLock} |
| ARGV[1] | internalLockLeaseTime | 30000ms |
| ARGV[2] | getLockName(threadId) | The UUID of the client splices the current thread ID |
| ARGV[3] | currentTime + threadWaitTime | Current time + 5000 |
| ARGV[4] | currentTime | The current time |
3.1.1 Locking for the first time
Call (‘lindex’, redisson_lock_queue:{anyLock}, 0) to pop up the first element in the queue. If there are no elements in the queue, If false is returned, the while(true) loop is immediately exited.
if (redis.call('exists', KEYS[1]) == 0) and ((redis.call('exists', KEYS[2]) == 0)
or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then
Copy the codeRedisson_lock_queue {anyLock} :{anyLock} :{anyLock} :{anyLock} :{anyLock} :{anyLock} :{anyLock} :{anyLock} :{anyLock} :{anyLock} :{anyLock} Zrem removes redisson_lock_timeout:{anyLock}, which is currently empty, sets the set assignment, and sets the default expiration time to 30s. Finally, a nil is returned. Through the above analysis of the re-entranceable lock, we know that when nil is returned, it represents the lock is successful. At this time, there will be a scheduling task watchdog to check the key every 10 seconds and renew the key.
redis.call('lpop', KEYS[2]);
redis.call('zrem', KEYS[3], ARGV[2]);
redis.call('hset', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
Copy the code3.1.2 Locking Client B
Call (‘lindex’, redisson_lock_queue:{anyLock}, 0); redisson_queue :{anyLock}, 0); Just exit the while loop and continue
local timeout = ttl + tonumber(ARGV[3]);
if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then
redis.call('rpush', KEYS[2], ARGV[2]);
end;
return ttl;
Copy the codeZadd redisSON_lock_TIMEOUT :{anyLock}, timeout, uuid+threadId, zadd redisson_lock_timeout:{anyLock}, timeout, uuid+threadId The main logic is to insert a piece of data into the set with a score of timeout, the value is the current thread ID of the current client’S UUID splicing, and then return the expiration time. From the previous analysis of the reentrant lock, we can know that when the return value is not nil, it means that the lock failed. At this point, it will enter the while(true) loop, interval to try to regain the lock.
3.1.3 Locking Client C
Now client B fails to lock and the relevant information is put into the queue, client C also obtains the lock
local firstThreadId = redis.call('lindex', KEYS[2], 0);
local ttl;
if firstThreadId ~= false and firstThreadId ~= ARGV[2] then
ttl = tonumber(redis.call('zscore', KEYS[3], firstThreadId)) - tonumber(ARGV[4]);
else
ttl = redis.call('pttl', KEYS[1]);
end;
Copy the codeUuidB +threadIdB = uuidB+threadIdB = uuidB+threadIdB = uuidB Subtract the current client C lock (current time + 5000ms), then perform client B lock section, add yourself to the queue and set, queue.
local timeout = ttl + tonumber(ARGV[3]);
if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then
redis.call('rpush', KEYS[2], ARGV[2]);
end;
return ttl;
Copy the code3.2 Reentrant locking
The core logic is to get the key, do the INCR increment, reset the lifetime.
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;
Copy the code3.3 Queue score refresh
Client B and client C keep trying to acquire the lock in the loop. When executing the above lua script, they are basically not satisfied and will execute to the following paragraph
"local firstThreadId = redis.call('lindex', KEYS[2], 0); " + local ttl; " + "if firstThreadId ~= false and firstThreadId ~= ARGV[2] then " + "ttl = tonumber(redis.call('zscore', KEYS[3], firstThreadId)) - tonumber(ARGV[4]);" + "else " + "ttl = redis.call('pttl', KEYS[1]);" + "end; " + "local timeout = ttl + tonumber(ARGV[3]);" + "if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then " + "redis.call('rpush', KEYS[2], ARGV[2]);" + "end; "+Copy the codeIn this case, we assume that client B first executes here, and it will execute PTTL to check the lock on client A and the remaining lifetime, then add the TTL to (current time + 5000) to calculate the timeout value, and then refresh the score value of client B through Zadd. In this case, client B has zadd for the second time, so the return value of this is 0, and client B will not be queued again.
In this case, the same logic applies to client C when it is their turn to execute, so their order in the queue set will not change and they will not add themselves to the queue multiple times.
3.4 Rearranging queues
If the client B because of various reasons, for a long time not to get locked again, led to the score has not been refreshed, the client then C attempts to acquire locks, access to the client from the queue B, but found that since the client B long time didn’t update time, lead to score value is less than the current time, that will be executed zrem, and lpop, Client B is removed from the sorted set and queued, now in a while loop. After the deletion, client C continues with the logic, retrieves itself from the queued, finds that its score is also less than the current time, and continues to zream, rPOP, remove itself, and at the end, Rejoin yourself in queues and collections.
At this point, client B, returning to normal, comes back and tries to lock the logic before going, rejoining itself. So now the order in the queue and the sort set has changed from where it started, and that’s queue reordering.
"while true do "
+ "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);"
+ "if firstThreadId2 == false then "
+ "break;"
+ "end; "
+ "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));"
+ "if timeout <= tonumber(ARGV[4]) then "
+ "redis.call('zrem', KEYS[3], firstThreadId2); "
+ "redis.call('lpop', KEYS[2]); "
+ "else "
+ "break;"
+ "end; "
+ "end;"
Copy the codeFrom this, we can see that after one client locks, other clients scramble for the lock. At first, within a certain period of time, but not too long, each client can keep orderly in the queue and collection at a fair pace.
Within a certain time frame, the time should not be too long, so that the order of the data in the queue and collection does not change, and each client periodically refreshes its score value.
However, if client A holds the lock for too long, some clients with too long waiting time may be deleted from the queue and set in the while true infinite loop. Once deleted, each client will reorder and join the queue and set in the time order of its own re-attempt to lock.
3.5 release the lock
When client A releases the lock, it also goes through the while True loop, looks at the timeout time of the elements in the ordered collection, and if it is less than the current time, deletes it and reorders everything that follows.
In this case, client B and CLIENT C both use tryAcquire method when trying to acquire the lock, and there will be a timeout time for acquiring the lock. When the timeout time is exceeded, they will not try to acquire the lock again, but the data in the queue and collection still exist. So this while true is going to do the culling. Even if the client is down, the score is not refreshed and will be removed from the collection and queue sooner or later when executing while True.
The logic is to make some judgments, publish some messages, and, if all goes well, execute the del command to delete the key. Since it is reentrant, it will decrement by 1 to determine if it is 0, and reset the survival time if it is not 0.
"while true do " + "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" + "if firstThreadId2 == false then " + "break;" + "end; " + "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" + "if timeout <= tonumber(ARGV[4]) then " + "redis.call('zrem', KEYS[3], firstThreadId2); " + "redis.call('lpop', KEYS[2]); " + "else " + "break;" + "end; " + "end;" + "if (redis.call('exists', KEYS[1]) == 0) then " + "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + "if nextThreadId ~= false then " + "redis.call('publish', KEYS[4] .. ':'.. nextThreadId, ARGV[1]); " + "end; " + "return 1; " + "end;" + "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; " + "end; " + "redis.call('del', KEYS[1]); " + "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + "if nextThreadId ~= false then " + "redis.call('publish', KEYS[4] .. ':'.. nextThreadId, ARGV[1]); " + "end; " + "return 1; ",Copy the codeWhen client A releases the lock, then client C goes to lock logic, then the key is empty, and the queue head is client C, then client C will remove itself from the queue and set, and then set the lock key, set the timeout period, return nil, add watchdog.
Fourth, the MultiLock
Combine multiple locks into a large lock, apply for a large lock and release the lock uniformly, lock multiple resources at a time, and then process the task, and then release it at a time
4.1 Code Examples
public static void main(String[] args) { RLock lock1 = redissonInstance1.getLock("lock1"); RLock lock2 = redissonInstance2.getLock("lock2"); RLock lock3 = redissonInstance3.getLock("lock3"); RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3); Lock1 lock2 lock3 // All locks are successful. lock.lock(); . lock.unlock(); }Copy the code4.2 lock
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException { long baseWaitTime = locks.size() * 1500; long waitTime = -1; if (leaseTime == -1) { waitTime = baseWaitTime; unit = TimeUnit.MILLISECONDS; } else { waitTime = unit.toMillis(leaseTime); if (waitTime <= 2000) { waitTime = 2000; } else if (waitTime <= baseWaitTime) { waitTime = ThreadLocalRandom.current().nextLong(waitTime/2, waitTime); } else { waitTime = ThreadLocalRandom.current().nextLong(baseWaitTime, waitTime); } waitTime = unit.convert(waitTime, TimeUnit.MILLISECONDS); } while (true) { if (tryLock(waitTime, leaseTime, unit)) { return; }}}Copy the codeThe lock logic here is simple. First, it calculates a baseWaitTime (4500) based on the number of locks, and then enters an infinite loop while(true), using the tryLock() method to obtain the lock. TryLock () is used here, specifying a maximum wait time of 2000 for acquiring the lock and a default expiration time of 30,000 milliseconds.
* * * *
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException { long newLeaseTime = -1; if (leaseTime ! = -1) { newLeaseTime = unit.toMillis(waitTime)*2; } long time = System.currentTimeMillis(); long remainTime = -1; if (waitTime ! = -1) { remainTime = unit.toMillis(waitTime); } long lockWaitTime = calcLockWaitTime(remainTime); int failedLocksLimit = failedLocksLimit(); List<RLock> acquiredLocks = new ArrayList<RLock>(locks.size()); for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) { long awaitTime = Math.min(lockWaitTime, remainTime); lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS); if (lockAcquired) { acquiredLocks.add(lock); } else { ... . } // Calculate time if (remainTime! = -1) { remainTime -= (System.currentTimeMillis() - time); time = System.currentTimeMillis(); If (remainTime <= 0) {// unlockInner(acquiredLocks); return false; }}}... . return true; }Copy the codeIn simple terms, we iterate over the set of locks we’ve set, iterating to try to get the lock, and after each lock is acquired, we count the remaining locks, and if we don’t get all the locks within 4500 milliseconds, we release the locks that we’ve already acquired. It then returns false and, in an infinite loop while(true), continues execution.
4.3 release the lock
public void unlock() { List<RFuture<Void>> futures = new ArrayList<RFuture<Void>>(locks.size()); for (RLock lock : locks) { futures.add(lock.unlockAsync()); } for (RFuture<Void> future : futures) { future.syncUninterruptibly(); }}Copy the codeThe lock release logic is simple: iterate over the lock collection, call the Lua script to release the lock, and then have a future waiting to complete.
Fifth, RedLock
5.1 the principle
If we want to acquire a distributed lock in cluster mode, we need to go through the following steps:
- Get the current timestamp at the time of execution;
- Try to create a lock on each master, set a short expiration time, usually tens of milliseconds, during the lock creation process, set a timeout period, if the lock did not obtain the success of the timeout period, as failure.
- Try to create locks on most nodes (n / 2 + 1);
- The client calculates the lock creation time. If the lock creation time is shorter than the timeout period, the lock is created successfully.
- If the lock creation fails, delete the lock that has been created.
- As soon as someone else creates a distributed lock, you have to constantly train yourself to try to accidentally unlock it
The common Redis distributed lock is actually created by selecting an instance through the hash algorithm, but the RedLock needs to be successfully created on N / 2 + 1 nodes to be regarded as the overall success, avoiding the lock on only one node instance.
5.2 Algorithm Implementation
public static void main(String[] args) { RLock lock1 = redissonInstance1.getLock("lock1"); RLock lock2 = redissonInstance2.getLock("lock2"); RLock lock3 = redissonInstance3.getLock("lock3"); RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3); // Lock at the same time: lock1 lock2 lock3 // Red lock On most nodes if the lock is successful. lock.lock(); . lock.unlock(); } public class RedissonRedLock extends RedissonMultiLock { /** * Creates instance with multiple {@link RLock} objects. * Each RLock object could be created by own Redisson instance. * * @param locks - array of locks */ public RedissonRedLock(RLock... locks) { super(locks); } @Override protected int failedLocksLimit() { return locks.size() - minLocksAmount(locks); } protected int minLocksAmount(final List<RLock> locks) { return locks.size()/2 + 1; } @Override protected long calcLockWaitTime(long remainTime) { return Math.max(remainTime / locks.size(), 1); } @Override public void unlock() { unlockInner(locks); }}Copy the codeIn fact, the redLock class is a subclass of MultiLock, it overwrites the related parameters of a calculation logic, here we play three lock calculation
-
- FailedLocksLimit: lock.size – lock.size() / 2 + 1 = 1, which indicates the number of allowed failures.
- CalcLockWaitTime: remainTime/lock.size() = 4500/3 = 1500, obtain the maximum wait time for each small lock
for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) { RLock lock = iterator.next(); boolean lockAcquired; lockAcquired = lock.tryLock(); . . if (lockAcquired) { acquiredLocks.add(lock); } else { if (locks.size() - acquiredLocks.size() == failedLocksLimit()) { break; } if (failedLocksLimit == 0) { unlockInner(acquiredLocks); if (waitTime == -1 && leaseTime == -1) { return false; } failedLocksLimit = failedLocksLimit(); acquiredLocks.clear(); // reset iterator while (iterator.hasPrevious()) { iterator.previous(); } } else { failedLocksLimit--; } } if (remainTime ! = -1) { remainTime -= (System.currentTimeMillis() - time); time = System.currentTimeMillis(); if (remainTime <= 0) { unlockInner(acquiredLocks); return false; }}}}Copy the codeLock. Size – The length of the lock set == the number of failed attempts. If this condition is met, break the for loop and return true. The lock is successfully added. If not, the number of allowed failures — is executed until the number reaches 0, the locks that have been added are released, and false is returned indicating that the whole lock failed.
There are three lock keys distributed on three different instances of Redis Master. If someone else tries to lock the redis master with the same key, the lock cannot be successfully locked because the lock has already been occupied, so it will enter a while(true) loop to try to acquire the lock. The logic is the same as before.
Read/write lock
6.1 read lock
"local mode = redis.call('hget', KEYS[1], 'mode'); " + "if (mode == false) then " + "redis.call('hset', KEYS[1], 'mode', 'read'); " + "redis.call('hset', KEYS[1], ARGV[2], 1); " + "redis.call('set', KEYS[2] .. : '1', 1); " + "redis.call('pexpire', KEYS[2] .. ':1', ARGV[1]); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "if (mode == 'read') or (mode == 'write' and redis.call('hexists', KEYS[1], ARGV[3]) == 1) then " + "local ind = redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "local key = KEYS[2] .. ':'.. ind;" + "redis.call('set', key, 1); " + "redis.call('pexpire', key, ARGV[1]); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end;" + "return redis.call('pttl', KEYS[1]);" , Arrays.<Object>asList(getName(), getReadWriteTimeoutNamePrefix(threadId)), internalLockLeaseTime, getLockName(threadId), getWriteLockName(threadId));Copy the codeLock is mainly in the execution of this script, first we analyze the parameters:
-
- KEYS[1] : Currently set lock Key = anyLock
- KEYS[2] : {lock key}: client UUID :threadId:rwlock_timeout = {anyLock}:UUID_01:ThreadId_01:rwlock_timeout
- ARGV[1] : Default expiration time = 30000ms
- ARGV[2] : client UUID:threadID = UUID_01:ThreadId_01
- ARGV[3] : client UUID:threadID:write = UUID_01:ThreadId_01:write
If anyLock does not exist, it returns false and executes the following statement:
Hset anyLock mode read, set a hash value
Hset anyLock UUID_01:ThreadId_01 1. Set the key to 1
Set {anyLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1 Set the key to 1
Pexpire {anyLock}:UUID_01:ThreadId_01:rwlock_timeout 30000 Sets the expiration time of 30s
Pexpire anyLock 30000 Sets the expiration time of 30s
If the lock is successful, then the logic is the same as before, if the lock is successful, it will trigger a scheduling task, add watchdog, every 10 seconds, to check if the key is still occupied, refresh the lifetime.
"local counter = redis.call('hget', KEYS[1], ARGV[2]); " +
"if (counter ~= false) then " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"if (redis.call('hlen', KEYS[1]) > 1) then " +
"local keys = redis.call('hkeys', KEYS[1]); " +
"for n, key in ipairs(keys) do " +
"counter = tonumber(redis.call('hget', KEYS[1], key)); " +
"if type(counter) == 'number' then " +
"for i=counter, 1, -1 do " +
"redis.call('pexpire', KEYS[2] .. ':' .. key .. ':rwlock_timeout:' .. i, ARGV[1]); " +
"end; " +
"end; " +
"end; " +
"end; " +
"return 1; " +
"end; " +
"return 0;",
Arrays.<Object>asList(getName(), keyPrefix),
internalLockLeaseTime, getLockName(threadId));
Copy the codeAgain, let’s analyze what each parameter means:
-
- KEYS[1] : lock key = anyLock
- KEYS[2] : {anyLock}
- ARGV[1] : the default expiration time is 30000ms
- ARGV[2] : client UUID:ThreadId = UUID_01:ThreadId_01
The first is to get the value of the lock client in the hash structure when the lock is added, and determine if there is a value, directly refresh the anyLock value to the default 30 seconds.
Then check whether anyLock has multiple keys in the hash structure, which is satisfied, obtain all keys, traverse, find the key whose value is of type number, **hset anyLock UUID_01:ThreadId_01 1 ** Hset anyLock UUID_01:ThreadId_01 1 Pexpire {anyLock}:anyLock UUID_01:ThreadId_01:rwlock_timeout:1 30000ms Set {anyLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1. If 1 is returned, the refresh is successful.
6.2 write lock
"local mode = redis.call('hget', KEYS[1], 'mode'); " + "if (mode == false) then " + "redis.call('hset', KEYS[1], 'mode', 'write'); " + "redis.call('hset', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "if (mode == 'write') then " + "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "local currentExpire = redis.call('pttl', KEYS[1]); " + "redis.call('pexpire', KEYS[1], currentExpire + ARGV[1]); " + "return nil; " + "end; " + "end;" + "return redis.call('pttl', KEYS[1]);" , Arrays.<Object>asList(getName()), internalLockLeaseTime, getLockName(threadId));Copy the codeKEYS [1] = anyLock,
ARGV[1] = 30000ms , ARGV[2] = UUID_01:ThreadId_01:write
First, we get the value from mode, which is not available by default, so we do the following: set anyLock mode write,
Set anyLock UUID_01:ThreadId_01:write 1, then set the default expiration time.
6.3 Read Locks Read locks are not mutually exclusive
"if (mode == 'read') or (mode == 'write' and redis.call('hexists', KEYS[1], ARGV[3]) == 1) then " + "local ind = redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "local key = KEYS[2] .. ':'.. ind;" + "redis.call('set', key, 1); " + "redis.call('pexpire', key, ARGV[1]); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end;" + "return redis.call('pttl', KEYS[1]);" .Copy the code- KEYS[1] : Currently set lock Key = anyLock
- KEYS[2] : {lock key}: client uUID :threadId:rwlock_timeout = {anyLock}:UUID_02:ThreadId_02:rwlock_timeout
- ARGV[1] : Default expiration time = 30000ms
- ARGV[2] : client UUID:threadID = UUID_02:ThreadId_02
- ARGV[3] : client UUID:threadID:write = UUID_02:ThreadId_02:write
UUID_02:ThreadId_02 = UUID_02:ThreadId_02 = UUID_02:ThreadId_02 = UUID_02 {anyLock}:UUID_02:ThreadId_02:rwlock_timeout:1, set the key value to 1, and set the expiration time, refresh the primary key value.
When the lock is successfully locked, the scheduling task, also known as watchdog, is set up to perform the logic of the refresh lifecycle, which is the same as the previous analysis.
Read lock and read lock are not mutually exclusive. The key value will be added to anyLock’s primary key and the current client value will be set.
6.4 Read Locks Write locks are mutually exclusive
In fact, the lua script above can be found, if the read lock is added before the write lock is added, it will not walk any if, and finally execute the TTL command, return a remaining time, then it represents the lock failed, the client will enter a while(true) loop, trying to obtain the lock.
When the write lock is added first and then the read lock is added, only the write lock added by the current client can enter if and set the corresponding key information. Otherwise, the lock fails.