>>>

工作中需要寫一個定時任務,由於是叢集環境,自然而然想到需要通過分散式鎖來保證單臺執行..相信大家都會想到使用zk來實現對應的分散式鎖.下面就簡單介紹一下幾種實現

準備工作

有幾個幫助類,先把程式碼放上來

ZKClient 對zk的操作做了一個簡單的封裝

Java程式碼

1. packagezk.lock;
3. importorg.apache.zookeeper.*;
4. importorg.apache.zookeeper.data.Stat;
5. importzk.util.ZKUtil;
7. importjava.util.concurrent.CountDownLatch;
8. importjava.util.concurrent.TimeUnit;
10. /**
11. *User:zhenghui
12. *Date:14-3-26
13. *Time:下午8:50
14. *封裝一個zookeeper例項.
15. */
16. publicclassZKClientimplementsWatcher{
18. privateZooKeeperzookeeper;
20. privateCountDownLatchconnectedSemaphore=newCountDownLatch(1);
23. publicZKClient(StringconnectString,intsessionTimeout)throwsException{
24. zookeeper=newZooKeeper(connectString,sessionTimeout,this);
25. System.out.println("connectingzkserver");
26. if(connectedSemaphore.await(10l,TimeUnit.SECONDS)){
27. System.out.println("connectzkserversuccess");
28. }else{
29. System.out.println("connectzkservererror.");
30. thrownewException("connectzkservererror.");
31. }
32. }
34. publicvoidclose()throwsInterruptedException{
35. if(zookeeper!=null){
36. zookeeper.close();
37. }
38. }
40. publicvoidcreatePathIfAbsent(Stringpath,booleanisPersistent)throwsException{
41. CreateModecreateMode=isPersistent?CreateMode.PERSISTENT:CreateMode.EPHEMERAL;
42. path=ZKUtil.normalize(path);
43. if(!this.exists(path)){
44. zookeeper.create(path,null,ZooDefs.Ids.OPEN_ACL_UNSAFE,createMode);
45. }
46. }
48. publicbooleanexists(Stringpath)throwsException{
49. path=ZKUtil.normalize(path);
50. Statstat=zookeeper.exists(path,null);
51. returnstat!=null;
52. }
54. publicStringgetData(Stringpath)throwsException{
55. path=ZKUtil.normalize(path);
56. try{
57. byte[]data=zookeeper.getData(path,null,null);
58. returnnewString(data);
59. }catch(KeeperExceptione){
60. if(einstanceofKeeperException.NoNodeException){
61. thrownewException("Nodedoesnotexist,pathis["+e.getPath()+"].",e);
62. }else{
63. thrownewException(e);
64. }
65. }catch(InterruptedExceptione){
66. Thread.currentThread().interrupt();
67. thrownewException(e);
68. }
69. }
71. @Override
72. publicvoidprocess(WatchedEventevent){
73. if(event==null)return;
75. //連線狀態
76. Watcher.Event.KeeperStatekeeperState=event.getState();
77. //事件型別
78. Watcher.Event.EventTypeeventType=event.getType();
79. //受影響的path
80. //Stringpath=event.getPath();
81. if(Watcher.Event.KeeperState.SyncConnected==keeperState){
82. //成功連線上ZK伺服器
83. if(Watcher.Event.EventType.None==eventType){
84. System.out.println("zookeeperconnectsuccess");
85. connectedSemaphore.countDown();
86. }
87. }
88. //下面可以做一些重連的工作.
89. elseif(Watcher.Event.KeeperState.Disconnected==keeperState){
90. System.out.println("zookeeperDisconnected");
91. }elseif(Watcher.Event.KeeperState.AuthFailed==keeperState){
92. System.out.println("zookeeperAuthFailed");
93. }elseif(Watcher.Event.KeeperState.Expired==keeperState){
94. System.out.println("zookeeperExpired");
95. }
96. }
97. }

ZKUtil 針對zk路徑的一個工具類

Java程式碼

1. packagezk.util;
3. /**
4. *User:zhenghui
5. *Date:14-3-26
6. *Time:下午9:56
7. */
8. publicclassZKUtil{
10. publicstaticfinalStringSEPARATOR="/";
12. /**
13. *轉換path為zk的標準路徑以/開頭,最後不帶/
14. */
15. publicstaticStringnormalize(Stringpath){
16. Stringtemp=path;
17. if(!path.startsWith(SEPARATOR)){
18. temp=SEPARATOR+path;
19. }
20. if(path.endsWith(SEPARATOR)){
21. temp=temp.substring(0,temp.length()-1);
22. returnnormalize(temp);
23. }else{
24. returntemp;
25. }
26. }
28. /**
29. *連結兩個path,並轉化為zk的標準路徑
30. */
31. publicstaticStringcontact(Stringpath1,Stringpath2){
32. if(path2.startsWith(SEPARATOR)){
33. path2=path2.substring(1);
34. }
35. if(path1.endsWith(SEPARATOR)){
36. returnnormalize(path1+path2);
37. }else{
38. returnnormalize(path1+SEPARATOR+path2);
39. }
40. }
42. /**
43. *字串轉化成byte型別
44. */
45. publicstaticbyte[]toBytes(Stringdata){
46. if(data==null||data.trim().equals(""))returnnull;
47. returndata.getBytes();
48. }
49. }

NetworkUtil 獲取本機IP的工具方法

Java程式碼

1. packagezk.util;
3. importjava.net.InetAddress;
4. importjava.net.NetworkInterface;
5. importjava.util.Enumeration;
7. /**
8. *User:zhenghui
9. *Date:14-4-1
10. *Time:下午4:47
11. */
12. publicclassNetworkUtil{
14. staticprivatefinalcharCOLON=':';
16. /**
17. *獲取當前機器ip地址
18. *據說多網絡卡的時候會有問題.
19. */
20. publicstaticStringgetNetworkAddress(){
21. Enumeration<NetworkInterface>netInterfaces;
22. try{
23. netInterfaces=NetworkInterface.getNetworkInterfaces();
24. InetAddressip;
25. while(netInterfaces.hasMoreElements()){
26. NetworkInterfaceni=netInterfaces
27. .nextElement();
28. Enumeration<InetAddress>addresses=ni.getInetAddresses();
29. while(addresses.hasMoreElements()){
30. ip=addresses.nextElement();
31. if(!ip.isLoopbackAddress()
32. &&ip.getHostAddress().indexOf(COLON)==-1){
33. returnip.getHostAddress();
34. }
35. }
36. }
37. return"";
38. }catch(Exceptione){
39. return"";
40. }
41. }
42. }

--------------------------- 正文開始 -----------------------------------

這種實現非常簡單,具體的流程如下

對應的實現如下

Java程式碼

1. packagezk.lock;
4. importzk.util.NetworkUtil;
5. importzk.util.ZKUtil;
7. /**
8. *User:zhenghui
9. *Date:14-3-26
10. *Time:下午8:37
11. *分散式鎖實現.
12. *
13. *這種實現的原理是,建立某一個任務的節點,比如/lock/tasckname然後獲取對應的值,如果是當前的Ip,那麼獲得鎖,如果不是,則沒獲得
14. *.如果該節點不存在,則建立該節點,並把改節點的值設定成當前的IP
15. */
16. publicclassDistributedLock01{
18. privateZKClientzkClient;
21. publicstaticfinalStringLOCK_ROOT="/lock";
22. privateStringlockName;
25. publicDistributedLock01(StringconnectString,intsessionTimeout,StringlockName)throwsException{
26. //先建立zk連結.
27. this.createConnection(connectString,sessionTimeout);
29. this.lockName=lockName;
30. }
32. publicbooleantryLock(){
33. Stringpath=ZKUtil.contact(LOCK_ROOT,lockName);
34. StringlocalIp=NetworkUtil.getNetworkAddress();
35. try{
36. if(zkClient.exists(path)){
37. StringownnerIp=zkClient.getData(path);
38. if(localIp.equals(ownnerIp)){
39. returntrue;
40. }
41. }else{
42. zkClient.createPathIfAbsent(path,false);
43. if(zkClient.exists(path)){
44. StringownnerIp=zkClient.getData(path);
45. if(localIp.equals(ownnerIp)){
46. returntrue;
47. }
48. }
49. }
50. }catch(Exceptione){
51. e.printStackTrace();
52. }
53. returnfalse;
54. }
57. /**
58. *建立zk連線
59. *
60. */
61. protectedvoidcreateConnection(StringconnectString,intsessionTimeout)throwsException{
62. if(zkClient!=null){
63. releaseConnection();
64. }
65. zkClient=newZKClient(connectString,sessionTimeout);
66. zkClient.createPathIfAbsent(LOCK_ROOT,true);
67. }
68. /**
69. *關閉ZK連線
70. */
71. protectedvoidreleaseConnection()throwsInterruptedException{
72. if(zkClient!=null){
73. zkClient.close();
74. }
75. }
77. }

總結

網上有很多文章,大家的方法大多數都是建立一個root根節點,每一個trylock的客戶端都會在root下建立一個EPHEMERAL_SEQUENTIAL 的子節點,同時設定root的child 變更watcher(為了避免羊群效應,可以只新增

前一個節點的變更通知) .如果建立的節點的序號是最小,則獲取到鎖,否則繼續等待root的child 變更

框架簡介：

本系統一款通用的SOA中介軟體平臺，用來開發各類J2EE企業級應用，節省時間和人力成本。本系統採用MVC模式、AOP引擎、任務排程器、工作流、Ajax、攔截器、過濾器、快取、日誌監控、資料訪問、表示式、國際化等技術。

使用者許可權系統：
組織結構：角色、使用者、使用者組、組織機構；許可權點：頁面、方法、按鈕、資料許可權、分級授權

專案管理新體驗：
快速出原型系統、元件樹、版本控制、模組移植、協同開發、實時監控、釋出管理

可持續整合：
所有元件可移植、可定製、可擴充，開發成果不斷積累，形成可持續發展的良性迴圈

框架/平臺構成：
Springmvc + Mybatis + Shiro（許可權）+SSO（單點登入） + Tiles（模板） +ActiveMQ（訊息佇列） + Rest（服務） + WebService（服務）+ EHcache（快取） + Lucene（搜尋引擎） + Quartz（定時排程）+ Html5（支援PC、IOS、Android）

支援平臺平臺：
Windows XP、Windows 7 、Windows 10 、 Linux 、 Unix

伺服器容器：
Tomcat 5/6/7 、Jetty、JBoss、WebSphere 8.5

專案原始碼結構截圖：

專案執行截圖：

轉載於:https://my.oschina.net/grthrj/blog/729114