zookeeper入门之curator框架--几种锁的操作

灰太狼 2021-09-11 06:30 386阅读 0赞

**\[java\]**  [view plain][]   [copy][view plain]

[![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **package** com.git.zookeeper.passwordmanager.lock;  
2.    
3.  **import** java.util.ArrayList;  
4.  **import** java.util.List;  
5.  **import** java.util.concurrent.TimeUnit;  
6.    
7.  **import** org.apache.commons.lang.math.RandomUtils;  
8.  **import** org.apache.curator.RetryPolicy;  
9.  **import** org.apache.curator.framework.CuratorFramework;  
10. **import** org.apache.curator.framework.CuratorFrameworkFactory;  
11. **import** org.apache.curator.framework.api.ACLProvider;  
12. **import** org.apache.curator.framework.recipes.locks.InterProcessLock;  
13. **import** org.apache.curator.framework.recipes.locks.InterProcessMutex;  
14. **import** org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;  
15. **import** org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;  
16. **import** org.apache.curator.framework.recipes.locks.Lease;  
17. **import** org.apache.curator.framework.recipes.shared.SharedCount;  
18. **import** org.apache.curator.retry.RetryNTimes;  
19. **import** org.apache.zookeeper.ZooDefs.Perms;  
20. **import** org.apache.zookeeper.data.ACL;  
21. **import** org.apache.zookeeper.data.Id;  
22.   
23. /\*\* 
24.  \* zookeeper 锁功能的演示 基于curator 
25.  \*  
26.  \* @author songqinghu 
27.  \* 
28.  \*共享锁: 全局同步分布式锁, 同一时间两台机器只有一台能获得同一把锁.  
29.  \*共享读写锁: 用于分布式的读写互斥处理, 同时生成两个锁:一个读锁, 一个写锁, 
30.  \*        读锁能被多个应用持有, 而写锁只能一个独占, 当写锁未被持有时, 多个读锁持有者可以同时进行读操作  
31.  \*共享信号量: 在分布式系统中的各个JVM使用同一个zk lock path, 
32.  \*      该path将跟一个给定数量的租约(lease)相关联, 然后各个应用根据请求顺序获得对应的lease,  
33.  \*      相对来说, 这是最公平的锁服务使用方式.  
34.  \*多共享锁:内部构件多个共享锁(会跟一个znode path关联), 在acquire()过程中, 
35.  \*     执行所有共享锁的acquire()方法, 如果中间出现一个失败, 则将释放所有已require的共享锁;  
36.  \*      执行release()方法时, 则执行内部多个共享锁的release方法(如果出现失败将忽略)  
37.  \*/  
38. **public** **class** ZookeeperLockDemo \{  
39.   
40.     **private** **static** CuratorFramework client ;  
41.       
42.     **public** **static** **void** main(String\[\] args) \{  
43.           
44.         //testSharedLock();  
45.        // testReadWriterLock();  
46.         **for** (**int** i = 0; i < 4; i++) \{  
47.             testSharedSemaphore(i);  
48.         \}  
49.     \}  
50.       
51.     /\* 
52.      \* 共享锁测试 
53.      \*/  
54.     **private** **static** **void** testSharedLock()\{  
55.         Thread t1 = **new** Thread("t1")\{  
56.             **public** **void** run()\{  
57.                 sharedLock();  
58.             \}  
59.         \};  
60.         Thread t2 = **new** Thread("t2")\{  
61.             **public** **void** run()\{  
62.                 sharedLock();  
63.             \}  
64.         \};  
65.         t1.start();  
66.         t2.start();  
67.     \}  
68.       
69.     /\*\* 
70.      \*  
71.      \* @描述：第一种锁: 共享锁 
72.      \* @return void 
73.      \* @exception 
74.      \* @createTime：2016年5月19日 
75.      \* @author: songqinghu 
76.      \* @throws Exception  
77.      \*/  
78.     **private** **static** **void** sharedLock()\{  
79.         CuratorFramework client = getClient();  
80.         //这两个都是共享锁  
81.        // new InterProcessMutex(client, path)  
82.        // new InterProcessSemaphoreMutex(client, path)  
83.           
84.         InterProcessMutex sharedLock = **new** InterProcessMutex(client, "/sharedlock");  
85.         **try** \{  
86.             //锁是否被获取到  
87.             //超时 不在进行操作  
88.               **if**(sharedLock.acquire(50, TimeUnit.MILLISECONDS))\{  
89.                 //sharedLock.acquire();  
90.                 System.out.println(Thread.currentThread().getName() + "  is  get the shared lock");  
91.                 Thread.sleep(100000);  
92.                 System.out.println(Thread.currentThread().getName() + "  is  release the shared lock");  
93.             \}  
94.               
95.         \} **catch** (Exception e) \{  
96.             //日志记录一下 超时说明 有锁 可以不在操作  
97.               
98.         \}**finally** \{  
99.             **try** \{  
100.                 System.out.println(Thread.currentThread().getName() + "  the flag is " + sharedLock.isAcquiredInThisProcess());  
101.                 **if**(sharedLock.isAcquiredInThisProcess())//判断是否持有锁 进而进行锁是否释放的操作  
102.                 sharedLock.release();  
103.             \} **catch** (Exception e) \{  
104.                 e.printStackTrace();  
105.             \}  
106.               
107.         \}  
108.     \}  
109.     /\*\* 
110.      \*  
111.      \* @描述：测试读写锁 
112.      \* @return void 
113.      \* @exception 
114.      \* @createTime：2016年5月19日 
115.      \* @author: songqinghu 
116.      \*/  
117.     **private** **static** **void** testReadWriterLock()\{  
118.         //创建多线程 循环进行锁的操作  
119.         CuratorFramework client = getClient();  
120.           
121.         InterProcessReadWriteLock readWriteLock = **new** InterProcessReadWriteLock(client, "/readwriter");  
122.        **final**  InterProcessMutex readLock = readWriteLock.readLock();  
123.        **final**  InterProcessMutex writeLock = readWriteLock.writeLock();  
124.           
125.         List<Thread> jobs = **new** ArrayList<Thread>();  
126.           
127.         **for** (**int** i = 0; i < 20; i++) \{  
128.             Thread t = **new** Thread("写锁  " + i)\{  
129.                   
130.                 **public** **void** run()\{  
131.                     readWriterLock(writeLock);  
132.                 \}  
133.             \};  
134.             jobs.add(t);  
135.         \}  
136.           
137.         **for** (**int** i = 0; i < 1; i++) \{  
138.             Thread t = **new** Thread("读锁  " + i)\{  
139.                   
140.                 **public** **void** run()\{  
141.                     readWriterLock(readLock);  
142.                 \}  
143.             \};  
144.             jobs.add(t);  
145.         \}  
146.           
147.         **for** (Thread thread : jobs) \{  
148.             thread.start();  
149.         \}  
150.           
151.     \}  
152.       
153.     /\*\* 
154.      \*  
155.      \* @描述：读写锁演示 
156.      \* @return void 
157.      \* @exception 
158.      \* @createTime：2016年5月19日 
159.      \* @author: songqinghu 
160.      \*/  
161.     **private** **static** **void** readWriterLock(InterProcessLock lock)\{  
162.         System.out.println(Thread.currentThread().getName()+"   进入任务 " + System.currentTimeMillis());  
163.         **try** \{  
164.             **if**(lock.acquire(20, TimeUnit.MILLISECONDS))\{  
165.                 //System.err.println(Thread.currentThread().getName() + " 等待超时  无法获取到锁");  
166.             //执行任务 --读取 或者写入  
167.                **int** time = RandomUtils.nextInt(150);  
168.                  
169.                System.out.println(Thread.currentThread().getName()+"   执行任务开始");  
170.                Thread.sleep(time);  
171.               System.out.println(Thread.currentThread().getName()+"   执行任务完毕");  
172.             \}**else**\{  
173.               System.err.println(Thread.currentThread().getName() + " 等待超时  无法获取到锁");   
174.             \}  
175.         \} **catch** (Exception e) \{  
176.             e.printStackTrace();  
177.         \}**finally**\{  
178.             **try** \{  
179.                 **if**(lock.isAcquiredInThisProcess())  
180.                     lock.release();  
181.             \} **catch** (Exception e2) \{  
182.               
183.             \}  
184.         \}  
185.           
186.     \}  
187.       
188.       
189.     **private** **static** **void** testSharedSemaphore(**final** **int** x)\{  
190.         CuratorFramework client = getClient();  
191.         **final** InterProcessSemaphoreV2 semaphoreV2 =   
192.                /// new InterProcessSemaphoreV2(client, "/sharedsemaphore", 50);  
193.                 **new** InterProcessSemaphoreV2(client, "/sharedsemaphore",  
194.                         **new** SharedCount(client, "/semaphore", 2));  
195.         List<Thread> jobs  = **new** ArrayList<Thread>();  
196.         **for** (**int** i = 0; i < 2; i++) \{  
197.             Thread thread = **new** Thread(x +"　　共享信息锁 " + i)\{  
198.                 **public** **void** run()\{  
199.                     sharedSemaphore(semaphoreV2);  
200.                 \}  
201.             \};  
202.             jobs.add(thread);  
203.         \}  
204.         **for** (Thread thread : jobs) \{  
205.             thread.start();  
206.         \}  
207.     \}  
208.       
209.       
210.     /\*\* 
211.      \* 共享信号量 
212.      \* 设置总的数量 -->分布式情况下的最大并行数量 
213.      \* 按照请求顺序进行 执行权的分配 
214.      \* 可以设置超时 不执行 也可以设置 直到获取执行权 执行 
215.      \*/  
216.     **private** **static** **void** sharedSemaphore(InterProcessSemaphoreV2 semaphoreV2)\{  
217.         //CuratorFramework client = getClient();  
218.        // new InterProcessSemaphoreV2(client, path, maxLeases)  
219.        // new InterProcessSemaphoreV2(client, path, count)  
220.           
221.         Lease lease = **null**;  
222.         **try** \{  
223.            // lease = semaphoreV2.acquire(10, TimeUnit.MILLISECONDS);  
224.             lease = semaphoreV2.acquire();  
225.             **if**(lease != **null**)\{  
226.                 System.out.println(Thread.currentThread().getName()+"   执行任务开始" + System.currentTimeMillis());  
227.                 //Thread.sleep(RandomUtils.nextInt(4000));  
228.                 Thread.sleep(1000);  
229.                 System.out.println(Thread.currentThread().getName()+"   执行任务完毕" + System.currentTimeMillis());  
230.             \}  
231.               
232.         \} **catch** (Exception e) \{  
233.             e.printStackTrace();  
234.         \}**finally** \{  
235.             **try** \{  
236.                 **if**(lease !=**null**)  
237.                     semaphoreV2.returnLease(lease);  
238.             \} **catch** (Exception e2) \{  
239.                 e2.printStackTrace();  
240.             \}  
241.               
242.               
243.         \}  
244.           
245.           
246.           
247.           
248.     \}  
249.       
250.       
251.       
252.     /\*\* 
253.      \*  
254.      \* @描述：获取连接 
255.      \* @return void 
256.      \* @exception 
257.      \* @createTime：2016年5月19日 
258.      \* @author: songqinghu 
259.      \*/  
260.     **private** **static** CuratorFramework getClient()\{  
261.         **if**(client ==**null**)\{  
262.               
263.             ACLProvider aclProvider = **new** ACLProvider() \{  
264.                 **private** List<ACL> acl;  
265.                 //初始化操作  
266.                 **private** **synchronized** **void** init()\{  
267.                     **if**(acl ==**null**)\{  
268.                         acl = **new** ArrayList<ACL>();  
269.                         acl.add(**new** ACL(Perms.ALL, **new** Id("auth","admin:admin")));  
270.                     \}  
271.                 \}  
272.                 @Override  
273.                 **public** List<ACL> getDefaultAcl() \{  
274.                     **if**(acl ==**null**)\{  
275.                         init();  
276.                     \}  
277.                     **return** **this**.acl;  
278.                 \}  
279.                 @Override  
280.                 **public** List<ACL> getAclForPath(String path) \{  
281.                     **if**(acl ==**null**)\{  
282.                         init();  
283.                     \}  
284.                     **return** **this**.acl;  
285.                 \}  
286.             \};  
287.             String scheme = "digest";  
288.             **byte**\[\] auth = "admin:admin".getBytes();  
289.             **int** connectionTimeoutMs =5000;  
290.             String connectString = "10.125.2.44:2181";  
291.             **byte**\[\] defaultData ="默认数据".getBytes();  
292.             **int** maxCloseWaitMs = 5000;  
293.             String namespace = "testlock";  
294.             RetryPolicy retryPolicy = **new** RetryNTimes(Integer.MAX\_VALUE, 5000);  
295.             CuratorFramework clientinit = CuratorFrameworkFactory.builder().aclProvider(aclProvider).authorization(scheme, auth)  
296.                     .connectionTimeoutMs(connectionTimeoutMs).connectString(connectString).  
297.                     defaultData(defaultData).maxCloseWaitMs(maxCloseWaitMs).namespace(namespace)  
298.                     .retryPolicy(retryPolicy).build();  
299.               
300.             clientinit.start();   
301.             client = clientinit;  
302.         \}  
303.         **return** client;  
304.     \}  
305.       
306.       
307.       
308.       
309.       
310.       
311. \}

![save_snippets.png][]

[view plain]: http://blog.csdn.net/sqh201030412/article/details/51456143#
[CODE]: https://code.csdn.net/assets/CODE_ico.png
[CODE_CODE]: https://code.csdn.net/snippets/1689670
[ico_fork.svg]: https://code.csdn.net/assets/ico_fork.svg
[ico_fork.svg 1]: https://code.csdn.net/snippets/1689670/fork
[save_snippets.png]: /images/20210527/b6192f84413843ed89a3eeba616bcb76.png