[Middleware] Zookeeper implements a distributed lock from 0 to 1
Distributed lock, in the actual business use is one of the more commonly used in the scene, and the realization of the distributed lock, common in addition to redis, is the realization of the zk, in front of a post and introduces the basic concept of zk use position, so if let’s remember the zk features to design a distributed lock, you can do?
I. Program design
1. Create a node
Zk has four types of nodes, and one of the most obvious strategies is to create a node, and whoever succeeds in creating a node will represent who holds the lock
This idea is similar to setnx in Redis, because only one session will be successfully created in zK, and the rest will throw existing exceptions
With temporary nodes, the node is deleted after the session is lost, which avoids the problem that all instances cannot hold the lock due to the exception of the instance holding the lock and not voluntarily releasing the lock
With this scheme, if I want to implement the logic of blocking lock acquisition, then one of the schemes needs to write a while(true) to keep retrying
while(true) {
if (tryLock(xxx)) return true;
else Thread.sleep(1000);
}
Copy the codeAnother strategy is to use event listening to register a trigger for node deletion when the node exists, so THAT I don’t need to retry myself. Take full advantage of zK’s features to implement asynchronous callbacks
public void lock(a) {
if (tryLock(path, new Watcher() {
@Override
public void process(WatchedEvent event) {
synchronized(path){ path.notify(); {}}}))return true;
}
synchronized(path) { path.wait(); }}Copy the codeSo what’s wrong with this implementation?
Every time a node changes, all will listen to the change, the benefit of unfair lock support; The disadvantage is that only one of the remaining wakeup instances will preempt the lock, wasting meaningless wakeup performance
2. Temporary sequential node mode
This scenario becomes more common later on, and is the case for most of the evening’s tutorials, where the idea is to create temporary order nodes
The lock preemption succeeds only for the node with the smallest serial number. If it is not the smallest node, then listen for the deletion event of the node in front of it. If the node in front of it is deleted, either it gives up the lock, or it releases the lock held by itself. In either case, for me, I need to scoop up all the nodes, or the lock is successfully taken. Or replace it with a front node
II. Distributed lock implementation
Let’s take a step-by-step look at how distributed locks can be implemented based on temporary sequential nodes
For ZK, we still use apache package ZooKeeper to operate; A distributed lock instance for Curator is provided subsequently
1. Rely on
The core depends on
<! -- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.7.0</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
Copy the codeVersion Description:
- Zk version: 3.6.2
- SpringBoot: 2.2.1. RELEASE
2. Simple distributed locks
The first step is all instance creation
public class ZkLock implements Watcher {
private ZooKeeper zooKeeper;
// Create a persistent node as the root of the distributed lock
private String root;
public ZkLock(String root) throws IOException {
try {
this.root = root;
zooKeeper = new ZooKeeper("127.0.0.1:2181".500 _000.this);
Stat stat = zooKeeper.exists(root, false);
if (stat == null) {
// Create one if it does not exist
createNode(root, true); }}catch(Exception e) { e.printStackTrace(); }}// Simple encapsulation node creation, where only persistent + temporary order is considered
private String createNode(String path, boolean persistent) throws Exception {
return zooKeeper.create(path, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, persistent ? CreateMode.PERSISTENT : CreateMode.EPHEMERAL_SEQUENTIAL); }}Copy the codeIn our design, we need to hold the current node and listen for changes to the previous node, so we add two members to the ZkLock instance
/** * The current node */
private String current;
/** * The previous node */
private String pre;
Copy the codeThe next step is to try to acquire the lock logic
- If current does not exist, create a temporary sequence node and assign current
- If current exists, it indicates that it has been created before and is waiting for the lock to be released
- Next, according to whether the current node order is the smallest, to indicate whether the lock is successfully held
- If the order is not the smallest, find the previous node and assign pre;
- Listen for pre changes
/** * Attempt to obtain the lock, create a sequential temporary node, if the minimum data, it indicates that the lock preemption success; Otherwise fail * *@return* /
public boolean tryLock(a) {
try {
String path = root + "/";
if (current == null) {
// Create a temporary sequence node
current = createNode(path, false);
}
List<String> list = zooKeeper.getChildren(root, false);
Collections.sort(list);
if (current.equalsIgnoreCase(path + list.get(0))) {
// Succeeded in obtaining the lock
return true;
} else {
// Failed to obtain the lock, find the previous node
int index = Collections.binarySearch(list, current.substring(path.length()));
// Query the one before the current node
pre = path + list.get(index - 1); }}catch (Exception e) {
e.printStackTrace();
}
return false;
}
Copy the codeTryLock does not listen for changes to the previous node. TryLock returns success or failure immediately, so there is no need to register the listener
Our listening logic is placed in lock() synchronous blocking
- An attempt to preempt the lock is returned on success
- If the lock fails, it listens for the deletion event of the previous node
public boolean lock(a) {
if (tryLock()) {
return true;
}
try {
// Listen for the deletion event of the previous node
Stat state = zooKeeper.exists(pre, true);
if(state ! =null) {
synchronized (pre) {
// Block and wait for the preceding node to be released
pre.wait();
// We will not return true directly, because the previous node was deleted not because it held the lock and released the lock. If the temporary node was deleted because the session was interrupted, we need to change the listening preNode
returnlock(); }}else {
// If the lock does not exist, try again
returnlock(); }}catch (Exception e) {
e.printStackTrace();
}
return false;
}
Copy the codeNote:
- Reinvoked when the node does not exist, or after the event triggers a callback
lock()That I hu Hansan again to compete for the lock?
Why not just return true? But need to compete again?
- Because the previous node is deleted, it may be caused by the session interruption of the previous node. But the lock is still in the hands of another instance. What I should do at this point is requeue
Finally, don’t forget to release the lock
public void unlock(a) {
try {
zooKeeper.delete(current, -1);
current = null;
zooKeeper.close();
} catch(Exception e) { e.printStackTrace(); }}Copy the codeAt this point, our distributed lock is complete, next we review the implementation process
- All the points are from the zK basics (create nodes, delete nodes, get all your points, listen for events).
- Lock Snatching = Create the node with the smallest serial number
- If the node is not the smallest, then listen for the previous node deletion event
This implementation supports lock reentrant (why? Because when the lock is not released, we save current, and when the current node exists, we directly judge whether it is the smallest. Instead of recreating)
3. The test
And then finally write a test case. Let’s see
@SpringBootApplication
public class Application {
private void tryLock(long time) {
ZkLock zkLock = null;
try {
zkLock = new ZkLock("/lock");
System.out.println("Try to acquire lock:" + Thread.currentThread() + " at: " + LocalDateTime.now());
boolean ans = zkLock.lock();
System.out.println("Execute business logic :" + Thread.currentThread() + " at:" + LocalDateTime.now());
Thread.sleep(time);
} catch (Exception e) {
e.printStackTrace();
} finally {
if(zkLock ! =null) { zkLock.unlock(); }}}public Application(a) throws IOException, InterruptedException {
new Thread(() -> tryLock(10 _000)).start();
Thread.sleep(1000);
// There is a 10s interval between acquiring the lock and executing it, because the thread above preempts the lock and holds it for 10s
new Thread(() -> tryLock(1 _000)).start();
System.out.println("---------over------------");
Scanner scanner = new Scanner(System.in);
String ans = scanner.next();
System.out.println("---> over --->" + ans);
}
public static void main(String[] args) { SpringApplication.run(Application.class); }}Copy the codeThe output is as follows
II. The other
0. Project
- Project: github.com/liuyueyi/sp…
- Project source: github.com/liuyueyi/sp…
1. An ashy Blog
As far as the letter is not as good, the above content is purely one’s opinion, due to the limited personal ability, it is inevitable that there are omissions and mistakes, if you find bugs or have better suggestions, welcome criticism and correction, don’t hesitate to appreciate
Below a gray personal blog, record all the study and work of the blog, welcome everyone to go to stroll
Wechat official account: One Grey Blog