ZooKeeper实现分布式锁

1.什么是分布式锁

一般的锁:一般我们说的锁是但进程多线程的锁,在多线程并发编程中,用于线程之间的数据同步,保护共享资源的访问

分布式锁:分布式锁指的是在分布式环境下,保护跨进程,跨主机,跨网络的共享资源,实现互斥访问,保证一致性

2.分布式锁的架构图

20160320214246073.png

3.分布式锁的算法流程

20160320221558695.png

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
package com.jike.lock;

import java.util.concurrent.TimeUnit;

public interface DistributedLock {

/*
* 获取锁,如果没有得到就等待
*/
public void acquire() throws Exception;

/*
* 获取锁,直到超时
*/
public boolean acquire(long time, TimeUnit unit) throws Exception;

/*
* 释放锁
*/
public void release() throws Exception;


}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;

public class BaseDistributedLock {

private final ZkClientExt client;
private final String path;

//zookeeper中locker节点的路径
private final String basePath;
private final String lockName;
private static final Integer MAX_RETRY_COUNT = 10;

public BaseDistributedLock(ZkClientExt client, String path, String lockName){

this.client = client;
this.basePath = path;
this.path = path.concat("/").concat(lockName);
this.lockName = lockName;

}

private void deleteOurPath(String ourPath) throws Exception{
client.delete(ourPath);
}

private String createLockNode(ZkClient client, String path) throws Exception{

return client.createEphemeralSequential(path, null);
}

private boolean waitToLock(long startMillis, Long millisToWait, String ourPath) throws Exception{

boolean haveTheLock = false;
boolean doDelete = false;

try
{

while ( !haveTheLock )
{
//获取lock节点下的所有节点
List<String> children = getSortedChildren();
String sequenceNodeName = ourPath.substring(basePath.length()+1);

//获取当前节点的在所有节点列表中的位置
int ourIndex = children.indexOf(sequenceNodeName);
//节点位置小于0,说明没有找到节点
if ( ourIndex<0 ){
throw new ZkNoNodeException("节点没有找到: " + sequenceNodeName);
}

//节点位置大于0说明还有其他节点在当前的节点前面,就需要等待其他的节点都释放
boolean isGetTheLock = ourIndex == 0;
String pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1);

if ( isGetTheLock ){

haveTheLock = true;

}else{
/**
* 获取当前节点的次小的节点,并监听节点的变化
*/
String previousSequencePath = basePath .concat( "/" ) .concat( pathToWatch );
final CountDownLatch latch = new CountDownLatch(1);
final IZkDataListener previousListener = new IZkDataListener() {

public void handleDataDeleted(String dataPath) throws Exception {
latch.countDown();
}

public void handleDataChange(String dataPath, Object data) throws Exception {
// ignore
}
};

try
{
//如果节点不存在会出现异常
client.subscribeDataChanges(previousSequencePath, previousListener);

if ( millisToWait != null )
{
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
if ( millisToWait <= 0 )
{
doDelete = true; // timed out - delete our node
break;
}

latch.await(millisToWait, TimeUnit.MICROSECONDS);
}
else
{
latch.await();
}
}
catch ( ZkNoNodeException e )
{
//ignore
}finally{
client.unsubscribeDataChanges(previousSequencePath, previousListener);
}

}
}
}
catch ( Exception e )
{
//发生异常需要删除节点
doDelete = true;
throw e;
}
finally
{
//如果需要删除节点
if ( doDelete )
{
deleteOurPath(ourPath);
}
}
return haveTheLock;
}

private String getLockNodeNumber(String str, String lockName)
{
int index = str.lastIndexOf(lockName);
if ( index >= 0 )
{
index += lockName.length();
return index <= str.length() ? str.substring(index) : "";
}
return str;
}

List<String> getSortedChildren() throws Exception
{
try{

List<String> children = client.getChildren(basePath);
Collections.sort
(
children,
new Comparator<String>()
{
public int compare(String lhs, String rhs)
{
return getLockNodeNumber(lhs, lockName).compareTo(getLockNodeNumber(rhs, lockName));
}
}
);
return children;

}catch(ZkNoNodeException e){

client.createPersistent(basePath, true);
return getSortedChildren();

}
}

protected void releaseLock(String lockPath) throws Exception{
deleteOurPath(lockPath);

}

/**
* 尝试获取锁
* @param time
* @param unit
* @return
* @throws Exception
*/
protected String attemptLock(long time, TimeUnit unit) throws Exception{

final long startMillis = System.currentTimeMillis();
final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;

String ourPath = null;
boolean hasTheLock = false;
boolean isDone = false;
int retryCount = 0;

//网络闪断需要重试一试
while ( !isDone )
{
isDone = true;

try
{
ourPath = createLockNode(client, path);
hasTheLock = waitToLock(startMillis, millisToWait, ourPath);
}
catch ( ZkNoNodeException e )
{
if ( retryCount++ < MAX_RETRY_COUNT )
{
isDone = false;
}
else
{
throw e;
}
}
}
if ( hasTheLock )
{
return ourPath;
}

return null;
}


}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer;

public class TestDistributedLock {

public static void main(String[] args) {

final ZkClientExt zkClientExt1 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
final SimpleDistributedLockMutex mutex1 = new SimpleDistributedLockMutex(zkClientExt1, "/Mutex");

final ZkClientExt zkClientExt2 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
final SimpleDistributedLockMutex mutex2 = new SimpleDistributedLockMutex(zkClientExt2, "/Mutex");

try {
mutex1.acquire();
System.out.println("Client1 locked");
Thread client2Thd = new Thread(new Runnable() {

public void run() {
try {
mutex2.acquire();
System.out.println("Client2 locked");
mutex2.release();
System.out.println("Client2 released lock");

} catch (Exception e) {
e.printStackTrace();
}
}
});
client2Thd.start();
Thread.sleep(5000);
mutex1.release();
System.out.println("Client1 released lock");

client2Thd.join();

} catch (Exception e) {

e.printStackTrace();
}

}

}

原文链接:https://blog.csdn.net/ZuoAnYinXiang/article/details/50938430

分享到