負載均衡zookpeer之輪詢演算法

1 Ngix 和 dubbo的負載均衡的區別？

Ngix負責前端的負載均衡，zookeeper負責後台的dubbo的rpc的集群的負載均衡

2 代碼實現

監聽者

2. import java.io.IOException;
4. import org.apache.log4j.Logger;
5. import org.apache.zookeeper.CreateMode;
6. import org.apache.zookeeper.KeeperException;
7. import org.apache.zookeeper.WatchedEvent;
8. import org.apache.zookeeper.Watcher;
9. import org.apache.zookeeper.ZooDefs.Ids;
10. import org.apache.zookeeper.ZooKeeper;
11. import org.apache.zookeeper.data.Stat;
13. public class WatchMore
14. {
15. /**
16. * Logger for this class
17. */
18. private static final Logger logger = Logger.getLogger(WatchMore.class);
20. // 定義常量
21. private static final String CONNECTSTRING = "192.168.10.167:2181";
22. private static final int SESSION_TIMEOUT = 50 * 1000;
23. private static final String PATH = "/atguigu";
24. // 定義實例變數
25. private ZooKeeper zk = null;
26. private String oldValue = "";
28. public ZooKeeper startZK() throws IOException
29. {
30. return new ZooKeeper(CONNECTSTRING, SESSION_TIMEOUT, new Watcher() {
31. @Override
32. public void process(WatchedEvent event)
33. {
34. }
35. });
37. }
39. public void createZNode(String nodePath, String nodeValue) throws KeeperException, InterruptedException
40. {
41. zk.create(nodePath, nodeValue.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
42. }
44. public String getZNode(String nodePath) throws KeeperException, InterruptedException
45. {
46. String result = "";
48. byte[] byteArray = zk.getData(nodePath, new Watcher() {
49. @Override
50. public void process(WatchedEvent event)
51. {
52. try {
53. triggerValue(nodePath);
54. }
55. catch (KeeperException | InterruptedException e) {
56. e.printStackTrace();
57. }
58. }
59. }, new Stat());
61. result = new String(byteArray);
62. oldValue = result;
64. return result;
65. }
67. public boolean triggerValue(String nodePath) throws KeeperException, InterruptedException
68. {
69. String result = "";
71. byte[] byteArray = zk.getData(nodePath, new Watcher() {
72. @Override
73. public void process(WatchedEvent event)
74. {
75. try {
76. triggerValue(nodePath);
77. }
78. catch (KeeperException | InterruptedException e) {
79. e.printStackTrace();
80. }
81. }
82. }, new Stat());
84. result = new String(byteArray);
86. String newValue = result;
88. if (oldValue.equals(newValue)) {
89. logger.info("**************no change****************");
90. return false;
91. }
92. else {
93. logger.info("**************oldValue:" + oldValue + " newValue: " + newValue);
94. oldValue = newValue;
95. return true;
96. }
97. }
99. public static void main(String[] args) throws IOException, KeeperException, InterruptedException
100. {
101. WatchMore myWatch = new WatchMore();
103. // 1 獲得zk實例
104. myWatch.setZk(myWatch.startZK());
106. if (myWatch.getZk().exists(PATH, false) == null)
107. {
108. String initValue = "AAA";
109. myWatch.createZNode(PATH, initValue);
111. logger.info("***************main: " + myWatch.getZNode(PATH));
113. Thread.sleep(Long.MAX_VALUE);
115. }else {
116. logger.info("***************have node*******");
117. }
118. }
120. // setter------getter
121. public ZooKeeper getZk()
122. {
123. return zk;
124. }
126. public void setZk(ZooKeeper zk)
127. {
128. this.zk = zk;
129. }
131. public String getOldValue()
132. {
133. return oldValue;
134. }
136. public void setOldValue(String oldValue)
137. {
138. this.oldValue = oldValue;
139. }
141. }

3 實現負載均衡的服務

2. import java.io.IOException;
3. import java.util.ArrayList;
4. import java.util.List;
7. import org.apache.log4j.Logger;
8. import org.apache.zookeeper.KeeperException;
9. import org.apache.zookeeper.WatchedEvent;
10. import org.apache.zookeeper.Watcher;
11. import org.apache.zookeeper.ZooKeeper;
12. import org.apache.zookeeper.data.Stat;
13. import com.atguigu.zk.watch.WatchMore;
14. public class BalanceTest
15. {
16. /**
17. * Logger for this class
18. */
19. private static final Logger logger = Logger.getLogger(WatchMore.class);
20. // 定義常量
21. private static final String CONNECTSTRING = "192.168.31.167:2181";
22. private static final int SESSION_TIMEOUT = 50 * 1000;
23. private static final String PATH = "/atcunzi";
24. private static final String SUB_PREFIX = "sub";
25. // 定義實例變數
26. private ZooKeeper zk = null;
27. private List<String> serviceLists = new ArrayList<String>(); // sub1,.....sub5 // 用來收集可用服務的
28. private int subCount = 5; // 可用服務個數
29. private int currentServiceNum = 0; // 當前請求序號

30. // 初始化zookeepr，並獲取服務的集合

31. public ZooKeeper startZK() throws IOException
32. {
33. return new ZooKeeper(CONNECTSTRING, SESSION_TIMEOUT,new Watcher() {
34. @Override
35. public void process(WatchedEvent event)
36. {
37. try
38. {
39. serviceLists = zk.getChildren(PATH,true);
40. }catch (KeeperException | InterruptedException e) {
41. e.printStackTrace();
42. }
43. }
44. });
45. }
47. // 處理輪詢演算法邏輯 重要的是怎麼能讓服務 輪詢
48. public String dealRquest() throws KeeperException, InterruptedException
49. {
50. currentServiceNum = currentServiceNum + 1;
52. for (int i = currentServiceNum; i <=subCount; i++)
53. {
54. if(serviceLists.contains(SUB_PREFIX+currentServiceNum))
55. {
56. return new String(zk.getData(PATH+"/"+SUB_PREFIX+currentServiceNum,false,new Stat()));
57. }else{
58. currentServiceNum = currentServiceNum + 1;
59. }
60. }
61. //重要的是怎麼能讓服務 輪詢 1 3 5 currentServiceNum = 6
62. for (int i = 1; i <=subCount; i++)
63. {
64. if(serviceLists.contains(SUB_PREFIX+i))
65. {
66. currentServiceNum = i;
67. return new String(zk.getData(PATH+"/"+SUB_PREFIX+currentServiceNum,false,new Stat()));
68. }
69. }
70. return "no subNode~~~~~";
71. }
73. public static void main(String[] args) throws IOException, InterruptedException
74. {
75. BalanceTest test = new BalanceTest();
77. test.setZk(test.startZK());
78. Thread.sleep(3000);
79. String result = null;
80. for (int i = 1; i <=15; i++)
81. {
82. try
83. {
84. result = test.dealRquest();
86. logger.info("*********customerID: "+i+" currentWindows:"+test.currentServiceNum+" "+result);
88. Thread.sleep(2000);
89. }
90. catch (KeeperException | InterruptedException e) {
91. e.printStackTrace();
92. }
94. }
95. }
96. ///----setter=-=getter
97. public ZooKeeper getZk()
98. {
99. return zk;
100. }
101. public void setZk(ZooKeeper zk)
102. {
103. this.zk = zk;
104. }
105. }

4 總結

負載均衡可以看成一個銀行，請求看成是顧客，後台服務看成是銀行櫃檯，大堂經理可以看成zookeeper

服務啟動的時候，需要先準備各個銀行櫃檯都準備好，當顧客來到時候，可以 通過大堂經理的指引，去到相應的櫃檯進行櫃檯處理。

難點：當一個櫃檯暫停服務時候，大唐經理怎麼知道的，並且怎麼分發處理，並且服務怎麼輪詢的？

5 maven配置

2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3. xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4. <modelVersion>4.0.0</modelVersion>
6. <groupId>com.atguigu.zk1014</groupId>
7. <artifactId>zk1014</artifactId>
8. <version>0.0.1-SNAPSHOT</version>
9. <packaging>jar</packaging>
11. <name>zk1014</name>
12. <url>http://maven.apache.org</url>
14. <properties>
15. <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
16. </properties>
18. <dependencies>
19. <!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
20. <dependency>
21. <groupId>com.101tec</groupId>
22. <artifactId>zkclient</artifactId>
23. <version>0.10</version>
24. </dependency>
25. <!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
26. <dependency>
27. <groupId>org.apache.zookeeper</groupId>
28. <artifactId>zookeeper</artifactId>
29. <version>3.4.9</version>
30. </dependency>
31. <dependency>
32. <groupId>log4j</groupId>
33. <artifactId>log4j</artifactId>
34. <version>1.2.17</version>
35. </dependency>
36. <dependency>
37. <groupId>junit</groupId>
38. <artifactId>junit</artifactId>
39. <version>4.9</version>
40. <scope>test</scope>
41. </dependency>
42. </dependencies>
43. </project>

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！