基于zookeeper的分布式锁实现 【转】

• 工作中需要写一个定时任务,由于是集群环境,自然而然想到需要通过分布式锁来保证单台执行..相信大家都会想到使用zk来实现对应的分布式锁.下面就简单介绍一下几种实现

  准备工作

  有几个帮助类,先把代码放上来

  ZKClient 对zk的操作做了一个简单的封装

  Java代码

  ZKUtil 针对zk路径的一个工具类

  Java代码

  NetworkUtil 获取本机IP的工具方法

  Java代码

  --------------------------- 正文开始  -----------------------------------

  这种实现非常简单,具体的流程如下

  
   对应的实现如下

  Java代码

  总结

  网上有很多文章,大家的方法大多数都是创建一个root根节点,每一个trylock的客户端都会在root下创建一个 EPHEMERAL_SEQUENTIAL 的子节点,同时设置root的child 变更watcher(为了避免羊群效应,可以只添加前一个节点的变更通知) .如果创建的节点的序号是最小,则获取到锁,否则继续等待root的child 变更

1. package zk.lock;
2. import zk.util.NetworkUtil;
3. import zk.util.ZKUtil;
4. /**
5. * User: zhenghui
6. * Date: 14-3-26
7. * Time: 下午8:37
8. * 分布式锁实现.
9. *
10. * 这种实现的原理是,创建某一个任务的节点,比如 /lock/tasckname 然后获取对应的值,如果是当前的Ip,那么获得锁,如果不是,则没获得
11. * .如果该节点不存在,则创建该节点,并把改节点的值设置成当前的IP
12. */
13. public class DistributedLock01 {
14. private ZKClient zkClient;
15. public static final String LOCK_ROOT = "/lock";
16. private String lockName;
17. public DistributedLock01(String connectString, int sessionTimeout,String lockName) throws Exception {
18. //先创建zk链接.
19. this.createConnection(connectString,sessionTimeout);
20. this.lockName = lockName;
21. }
22. public boolean tryLock(){
23. String path = ZKUtil.contact(LOCK_ROOT,lockName);
24. String localIp = NetworkUtil.getNetworkAddress();
25. try {
26. if(zkClient.exists(path)){
27. String ownnerIp = zkClient.getData(path);
28. if(localIp.equals(ownnerIp)){
29. return true;
30. }
31. } else {
32. zkClient.createPathIfAbsent(path,false);
33. if(zkClient.exists(path)){
34. String ownnerIp = zkClient.getData(path);
35. if(localIp.equals(ownnerIp)){
36. return true;
37. }
38. }
39. }
40. } catch (Exception e) {
41. e.printStackTrace();
42. }
43. return false;
44. }
45. /**
46. * 创建zk连接
47. *
48. */
49. protected void createConnection(String connectString, int sessionTimeout) throws Exception {
50. if(zkClient != null){
51. releaseConnection();
52. }
53. zkClient = new ZKClient(connectString,sessionTimeout);
54. zkClient.createPathIfAbsent(LOCK_ROOT,true);
55. }
56. /**
57. * 关闭ZK连接
58. */
59. protected void releaseConnection() throws InterruptedException {
60. if (zkClient != null) {
61. zkClient.close();
62. }
63. }
64. }

1. package zk.util;
2. import java.net.InetAddress;
3. import java.net.NetworkInterface;
4. import java.util.Enumeration;
5. /**
6. * User: zhenghui
7. * Date: 14-4-1
8. * Time: 下午4:47
9. */
10. public class NetworkUtil {
11. static private final char COLON = ‘:‘;
12. /**
13. * 获取当前机器ip地址
14. * 据说多网卡的时候会有问题.
15. */
16. public static String getNetworkAddress() {
17. Enumeration<NetworkInterface> netInterfaces;
18. try {
19. netInterfaces = NetworkInterface.getNetworkInterfaces();
20. InetAddress ip;
21. while (netInterfaces.hasMoreElements()) {
22. NetworkInterface ni = netInterfaces
23. .nextElement();
24. Enumeration<InetAddress> addresses=ni.getInetAddresses();
25. while(addresses.hasMoreElements()){
26. ip = addresses.nextElement();
27. if (!ip.isLoopbackAddress()
28. && ip.getHostAddress().indexOf(COLON) == -1) {
29. return ip.getHostAddress();
30. }
31. }
32. }
33. return "";
34. } catch (Exception e) {
35. return "";
36. }
37. }
38. }

1. package zk.util;
2. /**
3. * User: zhenghui
4. * Date: 14-3-26
5. * Time: 下午9:56
6. */
7. public class ZKUtil {
8. public static final String SEPARATOR = "/";
9. /**
10. * 转换path为zk的标准路径 以/开头,最后不带/
11. */
12. public static String normalize(String path) {
13. String temp = path;
14. if(!path.startsWith(SEPARATOR)) {
15. temp = SEPARATOR + path;
16. }
17. if(path.endsWith(SEPARATOR)) {
18. temp = temp.substring(0, temp.length()-1);
19. return normalize(temp);
20. }else {
21. return temp;
22. }
23. }
24. /**
25. * 链接两个path,并转化为zk的标准路径
26. */
27. public static String contact(String path1,String path2){
28. if(path2.startsWith(SEPARATOR)) {
29. path2 = path2.substring(1);
30. }
31. if(path1.endsWith(SEPARATOR)) {
32. return normalize(path1 + path2);
33. } else {
34. return normalize(path1 + SEPARATOR + path2);
35. }
36. }
37. /**
38. * 字符串转化成byte类型
39. */
40. public static byte[] toBytes(String data) {
41. if(data == null || data.trim().equals("")) return null;
42. return data.getBytes();
43. }
44. }

1. package zk.lock;
2. import org.apache.zookeeper.*;
3. import org.apache.zookeeper.data.Stat;
4. import zk.util.ZKUtil;
5. import java.util.concurrent.CountDownLatch;
6. import java.util.concurrent.TimeUnit;
7. /**
8. * User: zhenghui
9. * Date: 14-3-26
10. * Time: 下午8:50
11. * 封装一个zookeeper实例.
12. */
13. public class ZKClient implements Watcher {
14. private ZooKeeper zookeeper;
15. private CountDownLatch connectedSemaphore = new CountDownLatch(1);
16. public ZKClient(String connectString, int sessionTimeout) throws Exception {
17. zookeeper = new ZooKeeper(connectString, sessionTimeout, this);
18. System.out.println("connecting zk server");
19. if (connectedSemaphore.await(10l, TimeUnit.SECONDS)) {
20. System.out.println("connect zk server success");
21. } else {
22. System.out.println("connect zk server error.");
23. throw new Exception("connect zk server error.");
24. }
25. }
26. public void close() throws InterruptedException {
27. if (zookeeper != null) {
28. zookeeper.close();
29. }
30. }
31. public void createPathIfAbsent(String path, boolean isPersistent) throws Exception {
32. CreateMode createMode = isPersistent ? CreateMode.PERSISTENT : CreateMode.EPHEMERAL;
33. path = ZKUtil.normalize(path);
34. if (!this.exists(path)) {
35. zookeeper.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode);
36. }
37. }
38. public boolean exists(String path) throws Exception {
39. path = ZKUtil.normalize(path);
40. Stat stat = zookeeper.exists(path, null);
41. return stat != null;
42. }
43. public String getData(String path) throws Exception {
44. path = ZKUtil.normalize(path);
45. try {
46. byte[] data = zookeeper.getData(path, null, null);
47. return new String(data);
48. } catch (KeeperException e) {
49. if (e instanceof KeeperException.NoNodeException) {
50. throw new Exception("Node does not exist,path is [" + e.getPath() + "].", e);
51. } else {
52. throw new Exception(e);
53. }
54. } catch (InterruptedException e) {
55. Thread.currentThread().interrupt();
56. throw new Exception(e);
57. }
58. }
59. @Override
60. public void process(WatchedEvent event) {
61. if (event == null) return;
62. // 连接状态
63. Watcher.Event.KeeperState keeperState = event.getState();
64. // 事件类型
65. Watcher.Event.EventType eventType = event.getType();
66. // 受影响的path
67. //        String path = event.getPath();
68. if (Watcher.Event.KeeperState.SyncConnected == keeperState) {
69. // 成功连接上ZK服务器
70. if (Watcher.Event.EventType.None == eventType) {
71. System.out.println("zookeeper connect success");
72. connectedSemaphore.countDown();
73. }
74. }
75. //下面可以做一些重连的工作.
76. else if (Watcher.Event.KeeperState.Disconnected == keeperState) {
77. System.out.println("zookeeper Disconnected");
78. } else if (Watcher.Event.KeeperState.AuthFailed == keeperState) {
79. System.out.println("zookeeper AuthFailed");
80. } else if (Watcher.Event.KeeperState.Expired == keeperState) {
81. System.out.println("zookeeper Expired");
82. }
83. }
84. }

框架/平台构成：Maven+Springmvc + Mybatis + Shiro（权限）+ Tiles（模板） +ActiveMQ（消息队列） + Rest（服务） + WebService（服务）+ EHcache（缓存） + Quartz（定时调度）+ Html5（支持PC、IOS、Android）用户权限系统：组织结构：角色、用户、用户组、组织机构；权限点：页面、方法、按钮、数据权限、分级授权项目管理新体验：快速出原型系统、组件树、版本控制、模块移植、协同开发、实时监控、发布管理可持续集成：所有组件可移植、可定制、可扩充，开发成果不断积累，形成可持续发展的良性循环支持平台平台： Windows XP、Windows 7 、Windows 10 、 Linux 、 Unix服务器容器：Tomcat 5/6/7 、Jetty、JBoss、WebSphere 8.5