代码已上传到Gtihub，有兴趣的同学可以下载来看看（git版本号90275563ea2f9efc047e62427f6b9d472fe47b7f）:https://github.com/ylw-github/Zookeeper-Demo
使用Zookeeper实现负载均衡原理：
- 服务器端将启动的服务注册到
ZK注册中心上，采用临时节点。 - 客户端从zk节点上获取最新服务节点信息，本地使用负载均衡算法，随机分配服务器。
下面来开始讲解代码的实现
ZK负载均衡实现
传统模式
1.添加maven依赖：
<dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.6</version> </dependency> <dependency> <groupId>com.101tec</groupId> <artifactId>zkclient</artifactId> <version>0.8</version> </dependency>
2.创建Server服务端:
ZkServerScoekt服务：
package com.ylw.zookeeper.loadbalance; import java.io.IOException; import java.net.ServerSocket; import java.net.Socket; public class ZkServerScoekt implements Runnable { private int port = 18080; 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) { e2.printStackTrace(); } } } public static void main(String[] args) throws IOException { int port = 18080; ZkServerScoekt server = new ZkServerScoekt(port); Thread thread = new Thread(server); thread.start(); } }
ServerHandler:
package com.ylw.zookeeper.loadbalance; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.io.PrintWriter; import java.net.Socket; 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; } } } }
3.创建客户端ZkServerClient:
package com.ylw.zookeeper.loadbalance; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.io.PrintWriter; import java.net.Socket; import java.util.ArrayList; import java.util.List; 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(); } } } } }
4. 运行Server和Client
客户端控制台输入并发送：Hello ,I am YLW !
服务端接收：
使用Zookeeper
1.改造ZkServerScoekt
package com.ylw.zookeeper.loadbalance; import org.I0Itec.zkclient.ZkClient; import java.io.IOException; import java.net.ServerSocket; import java.net.Socket; public class ZkServerScoekt2 implements Runnable { private static int port = 18081; //socket 服务启动后的所使用的 端口号 public static void main(String[] args) throws IOException { ZkServerScoekt server = new ZkServerScoekt(port); //构造函数传入port regServer(); //去zk注册 Thread thread = new Thread(server); thread.start(); } public ZkServerScoekt2(int port) { this.port = port; } //注册服务 public static void regServer() { //1、建立zk连接 ZkClient zkClient = new ZkClient("192.168.162.131", 5000, 10000); //2.先创建父节点 String root = "/toov5"; if (!zkClient.exists(root)) { //如果父节点不存，直接创建父节点 zkClient.createPersistent(root); //持久节点 } //3、创建子节点 String nodeName = root + "/service_" + port; String nodeValue = "127.0.0.1:" + port; if (zkClient.exists(nodeName)) { //如果存在 直接删除掉 zkClient.delete(nodeName); } zkClient.createEphemeral(nodeName, "127.0.0.1:" + port); System.out.println("服务注册成功" + nodeName); } 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：
package com.ylw.zookeeper.loadbalance; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.io.PrintWriter; import java.net.Socket; import java.util.ArrayList; import java.util.List; public class ZkServerClient2 { public static List<String> listServer = new ArrayList<String>(); public static void main(String[] args) { initServer(); ZkServerClient2 client = new ZkServerClient2(); 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:18081"); //连接服务 放到list中 存放集群地址 做负载均衡用的 } // 获取当前server信息 public static String getServer() { return listServer.get(0); } public void send(String name) { String server = ZkServerClient2.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(); } } } } }
