使用Zookeeper实现负载均衡原理
2024-09-27 19:56:09
思路
使用Zookeeper实现负载均衡原理，服务器端将启动的服务注册到，zk注册中心上，采用临时节点。客户端从zk节点上获取最新服务节点信息，本地使用负载均衡算法，随机分配服务器。
创建项目工程
Maven依赖
<dependencies>
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.8</version>
</dependency>
</dependencies>
创建Server服务端
ZkServerScoekt服务
//##ServerScoekt服务端
public class ZkServerScoekt implements Runnable {
private int port = 18080; public static void main(String[] args) throws IOException {
int port = 18080;
ZkServerScoekt server = new ZkServerScoekt(port);
Thread thread = new Thread(server);
thread.start();
} public ZkServerScoekt(int port) {
this.port = port;
} public void run() {
ServerSocket serverSocket = null;
try {
serverSocket = new ServerSocket(port);
System.out.println("Server start port:" + port);
Socket socket = null;
while (true) {
socket = serverSocket.accept();
new Thread(new ServerHandler(socket)).start();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (serverSocket != null) {
serverSocket.close();
}
} catch (Exception e2) { }
}
} }
ZkServerClient
public class ZkServerClient {
public static List<String> listServer = new ArrayList<String>();
public static void main(String[] args) {
initServer();
ZkServerClient client= new ZkServerClient();
BufferedReader console = new BufferedReader(new InputStreamReader(System.in));
while (true) {
String name;
try {
name = console.readLine();
if ("exit".equals(name)) {
System.exit(0);
}
client.send(name);
} catch (IOException e) {
e.printStackTrace();
}
}
}
// 注册所有server
public static void initServer() {
listServer.clear();
listServer.add("127.0.0.1:18080");
}
// 获取当前server信息
public static String getServer() {
return listServer.get(0);
}
public void send(String name) {
String server = ZkServerClient.getServer();
String[] cfg = server.split(":");
Socket socket = null;
BufferedReader in = null;
PrintWriter out = null;
try {
socket = new Socket(cfg[0], Integer.parseInt(cfg[1]));
in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
out = new PrintWriter(socket.getOutputStream(), true);
out.println(name);
while (true) {
String resp = in.readLine();
if (resp == null)
break;
else if (resp.length() > 0) {
System.out.println("Receive : " + resp);
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
if (out != null) {
out.close();
}
if (in != null) {
try {
in.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if (socket != null) {
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
ServerHandler
public class ServerHandler implements Runnable {
private Socket socket;
public ServerHandler(Socket socket) {
this.socket = socket;
}
public void run() {
BufferedReader in = null;
PrintWriter out = null;
try {
in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
out = new PrintWriter(this.socket.getOutputStream(), true);
String body = null;
while (true) {
body = in.readLine();
if (body == null)
break;
System.out.println("Receive : " + body);
out.println("Hello, " + body);
}
} catch (Exception e) {
if (in != null) {
try {
in.close();
} catch (IOException e1) {
e1.printStackTrace();
}
}
if (out != null) {
out.close();
}
if (this.socket != null) {
try {
this.socket.close();
} catch (IOException e1) {
e1.printStackTrace();
}
this.socket = null;
}
}
}
}
改造ZkServerScoekt
public class ZkServerScoekt implements Runnable {
private static int port = 18081;
public static void main(String[] args) throws IOException {
ZkServerScoekt server = new ZkServerScoekt(port);
Thread thread = new Thread(server);
thread.start();
}
public ZkServerScoekt(int port) {
this.port = port;
}
public void regServer() {
// 向ZooKeeper注册当前服务器
ZkClient client = new ZkClient("127.0.0.1:2181", 60000, 1000);
String path = "/test/server" + port;
if (client.exists(path))
client.delete(path);
client.createEphemeral(path, "127.0.0.1:" + port);
}
public void run() {
ServerSocket serverSocket = null;
try {
serverSocket = new ServerSocket(port);
regServer();
System.out.println("Server start port:" + port);
Socket socket = null;
while (true) {
socket = serverSocket.accept();
new Thread(new ServerHandler(socket)).start();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (serverSocket != null) {
serverSocket.close();
}
} catch (Exception e2) {
}
}
}
}
改造ZkServerClientScoekt
public class ZkServerClient {
public static List<String> listServer = new ArrayList<String>();
public static void main(String[] args) {
initServer();
ZkServerClient client = new ZkServerClient();
BufferedReader console = new BufferedReader(new InputStreamReader(System.in));
while (true) {
String name;
try {
name = console.readLine();
if ("exit".equals(name)) {
System.exit(0);
}
client.send(name);
} catch (IOException e) {
e.printStackTrace();
}
}
}
// 注册所有server
public static void initServer() {
String path = "/test";
final ZkClient zkClient = new ZkClient("127.0.0.1:2181", 6000, 1000);
List<String> children = zkClient.getChildren(path);
for (String ipServer : children) {
listServer.add((String) zkClient.readData(path + "/" + ipServer));
}
System.out.println("####从注册中心获取服务信息####listServer:" + listServer.toString());
// 监听事件
zkClient.subscribeChildChanges(path, new IZkChildListener() {
public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
listServer.clear();
for (String ctPath : currentChilds) {
listServer.add(zkClient.readData(parentPath) + "/" + ctPath);
}
System.out.println("####节点事件监听发生变化### listServer:" + listServer.toString());
}
});
// listServer.clear();
// listServer.add("127.0.0.1:8080");
}
private static int reqestCount = 1;
// 获取当前server信息
public static String getServer() {
int serverCount = listServer.size();
String serverHost = listServer.get(reqestCount / serverCount);
reqestCount++;
return serverHost;
}
public void send(String name) {
String server = ZkServerClient.getServer();
String[] cfg = server.split(":");
Socket socket = null;
BufferedReader in = null;
PrintWriter out = null;
try {
socket = new Socket(cfg[0], Integer.parseInt(cfg[1]));
in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
out = new PrintWriter(socket.getOutputStream(), true);
out.println(name);
while (true) {
String resp = in.readLine();
if (resp == null)
break;
else if (resp.length() > 0) {
System.out.println("Receive : " + resp);
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
if (out != null) {
out.close();
}
if (in != null) {
try {
in.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if (socket != null) {
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
使用Zookeeper实现选举策略
场景
有一个向外提供的服务，服务必须7*24小时提供服务，不能有单点故障。所以采用集群的方式，采用master、slave的结构。一台主机多台备机。主机向外提供服务，备机负责监听主机的状态，一旦主机宕机，备机要迅速接代主机继续向外提供服务。从备机选择一台作为主机，就是master选举。
原理分析
右边三台主机会尝试创建master节点，谁创建成功了，就是master，向外提供。其他两台就是slave。
所有slave必须关注master的删除事件（临时节点，如果服务器宕机了，Zookeeper会自动把master节点删除）。如果master宕机了，会进行新一轮的master选举。本次我们主要关注master选举，服务注册、发现先不讨论。
使用Zookeeper原理
» 领导者（leader），负责进行投票的发起和决议，更新系统状态
» 学习者（learner），包括跟随者（follower）和观察者（observer），follower用于接受客户端请求并想客户端返回结果，在选主过程中参与投票
» Observer可以接受客户端连接，将写请求转发给leader，但observer不参加投票过程，只同步leader的状态，observer的目的是为了扩展系统，提高读取速度
» 客户端（client），请求发起方
• Zookeeper的核心是原子广播，这个机制保证了各个Server之间的同步。实现这个机制的协议叫做Zab协
议。Zab协议有两种模式，它们分别是恢复模式（选主）和广播模式（同步）。当服务启动或者在领导者
崩溃后，Zab就进入了恢复模式，当领导者被选举出来，且大多数Server完成了和leader的状态同步以后
，恢复模式就结束了。状态同步保证了leader和Server具有相同的系统状态。
• 为了保证事务的顺序一致性，zookeeper采用了递增的事务id号（zxid）来标识事务。所有的提议（
proposal）都在被提出的时候加上了zxid。实现中zxid是一个64位的数字，它高32位是epoch用来标识
leader关系是否改变，每次一个leader被选出来，它都会有一个新的epoch，标识当前属于那个leader的
统治时期。低32位用于递增计数。
• 每个Server在工作过程中有三种状态：
LOOKING：当前Server不知道leader是谁，正在搜寻
LEADING：当前Server即为选举出来的leader
FOLLOWING：leader已经选举出来，当前Server与之同步
最新文章
- PHP global 关键字
- Basic Virus's Infection & Variation [Python]
- python 的dict的update 方法的一点诡秘的行为
- java判断文件是否存在
- 【转】PowerShell入门（七）：管道——在命令行上编程
- 案例：TableLayout表格布局——迷你计算器
- 设计模式之工厂模式(Factory)
- [转载]hao123军事频道首页JQ焦点图
- poj 3080 Blue Jeans（水题 暴搜）
- js 动态计算折扣后总价格
- .net 调用Oracle.Data.Access 组件提供的用于批量操作的方法
- 面向对象继承实例(a如何继承b)
- Python迭代
- 判断文件的编码 python
- 介绍一下Spring Cloud简介
- Nginx 配置location root 转自https://blog.csdn.net/rofth/article/details/78581617
- Android导入第三方静态库.a编译成动态库.so
- （转）C语言中长度为0的数组
- jquery validate 多种使用方式
- 201709013工作日记--static理解 && abstract
热门文章
- 【C++】继承和组合的概念？什么时候用继承？什么时候用组合？
- 19 SpringMVC 拦截器
- SPFA + 链式前向星(详解)
- 了解HTML/CSS/JS/JQuery/ajax等前端知识
- C++工程师养成 每日一题（string使用）
- Delphi 开发微信公众平台 （三）- 获取微信服务器IP地址
- 读文件时出现这个错误 'utf-8' codec can't decode byte 0xba in position 21: invalid start byte
- Python pip版本升级
- VM配置Centos（第十三步分区设置）
- C++ Primer 第五版示例gcc源码