Zookeeper實現服務註冊/發現
阿新 • • 發佈：2020-06-09
what that?
Zookeeper在分散式開發中使用頻繁,但許多框架都對其進行了封裝,初學者可能無法較好的理解其工作原理,該文章演示了使用Zookeeper實現服務註冊,服務發現的簡單demo,希望能達到拋磚引玉的效果;
why need RegisterCenter?
之所以需要訪問註冊和服務發現是因為分散式系統中,服務之間需要相互呼叫,但若每個服務自己維護一份依賴的服務資訊的話,就顯得很麻煩,且自身維護的資料無法保證其實時性,當依賴的服務資訊發生變更時,無法及時獲取更新,解決方案就是引入一個註冊中心,服務提供方將自己的資訊寫入到註冊中心,服務使用方從註冊中心來獲取服務資訊; 如下圖:
**client**表示服務使用方,**server**表示服務提供方
實現的效果: 客戶端可自動發現服務資訊,當服務狀態發生變化時(上線,下線,更換地址),客戶端可以及時響應變化,效果如下圖:
[效果演示](https://gitee.com/yangyuanhuyyh/MDimage/raw/8c49c890597f3e766d8b71a750faf085ba285727/uPic/Jun-09-2020%2010-46-19.gif)
# 實現
1. 首先保證Zookeeper以安裝啟動,且可以正常訪問
2. 建立Maven專案並新增Zookeeper的Java客戶端依賴(注意版本號需>3.6)
```xml
org.apache.zookeeper
zookeeper
3.6.1
```
3. 編寫服務提供方
```java
package com.jerry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import java.io.IOException;
import java.io.InputStream;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Enumeration;
import static java.net.InetAddress.getLocalHost;
public class UserService {
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
new UserService().serving();
}
public void serving() throws IOException, KeeperException, InterruptedException {
//獲取本機ip地址
String ip = null;
Enumeration networkInterfaces = NetworkInterface.getNetworkInterfaces();
while (networkInterfaces.hasMoreElements()) {
NetworkInterface ni = (NetworkInterface) networkInterfaces.nextElement();
Enumeration nias = ni.getInetAddresses();
while (nias.hasMoreElements()) {
InetAddress ia = (InetAddress) nias.nextElement();
if (!ia.isLinkLocalAddress() && !ia.isLoopbackAddress() && ia instanceof Inet4Address) {
ip = ia.getHostAddress();
}
}
}
int port = 8988;
//啟動服務
ServerSocket socket = new ServerSocket(port);
System.out.println("伺服器已啟動...");
//註冊服務
serverRegister(ip, port);
//處理請求
clientHandler(socket);
}
private void clientHandler(ServerSocket socket) throws IOException {
while (true) {
Socket accept = socket.accept();
InputStream inputStream = accept.getInputStream();
byte[] barr = new byte[1024];
while (true) {
int size = inputStream.read(barr);
if (size == -1) {
//System.out.println("客戶端已關閉..");
accept.close();
break;
}
String s = new String(barr, 0, size);
//輸出客戶端訊息
System.out.println(accept.getInetAddress().getHostAddress() + ": " + s);
}
}
}
private void serverRegister(String ip, int port) throws IOException, KeeperException, InterruptedException {
//註冊服務
ZooKeeper zooKeeper = new ZooKeeper("10.211.55.4", 2181, null);
try {
ArrayList acl = new ArrayList<>();
acl.add(new ACL(31, ZooDefs.Ids.ANYONE_ID_UNSAFE));
zooKeeper.create("/userServer", (ip + ":" + port).getBytes(StandardCharsets.UTF_8), acl, CreateMode.EPHEMERAL);
System.out.println("服務釋出成功!");
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
throw e;
}
}
}
```
4. 編寫服務服務使用方
```java
package com.yyh;
import org.apache.zookeeper.*;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Scanner;
public class UserClient implements Watcher {
String node = "/userServer"; //服務資訊所在的節點 服務提供方和服務消費方一致
private ZooKeeper zooKeeper;
String server_ip;
int server_port;
public static void main(String[] args) throws Exception {
//開始服務監聽
UserClient userClient = new UserClient();
userClient.run();
//當訪問可用時與服務互動
Scanner scanner = new Scanner(System.in);
while (true){
System.out.println("輸入要傳送的資訊(e:退出)");
String text = scanner.next();
if (text.equals("e"))System.exit(-1);
if (userClient.server_ip == null){
System.err.println("沒有可用的服務...");
}else {
userClient.sendToServer(text);
}
}
}
private void run() throws Exception {
//連線zookeeper
zooKeeper = new ZooKeeper("10.211.55.4:2181", 3000, null);
//嘗試獲取服務資訊
getServerInfo();
//新增對服務資訊的永久監聽
zooKeeper.addWatch(node,this,AddWatchMode.PERSISTENT);
}
//獲取服務資訊
private void getServerInfo() {
try {
byte[] data = zooKeeper.getData(node, false, null);
String[] infos = new String(data).split(":");
server_ip = infos[0];
server_port = Integer.parseInt(infos[1]);
System.out.println("獲取服務資訊成功!");
System.out.println(server_ip+":"+ server_port);
} catch (KeeperException e) {
System.err.println("服務資訊不存在! 等待服務上線........");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//當節點狀態傳送變化時將執行該方法(通知處理)
@Override
public void process(WatchedEvent event) {
if (event.getPath().equals(node)) {
//根據具體邏輯處理不同的事件型別,此處只關心節點的建立刪除和更新
if (event.getType() == Event.EventType.NodeCreated) {
System.err.println("服務上線了");
getServerInfo();
} else if (event.getType() == Event.EventType.NodeDataChanged) {
System.err.println("服務更新了");
getServerInfo();
}else if (event.getType()== Event.EventType.NodeDeleted){
server_ip = null;
server_port = 0;
System.err.println("服務下線了");
}
}
}
public void sendToServer(String text) {
InetSocketAddress server_address = new InetSocketAddress(server_ip, server_port);
Socket socket = new Socket();
try {
socket.connect(server_address);
//System.out.println("連線伺服器成功!");
OutputStream outputStream = socket.getOutputStream();
outputStream.write(text.getBytes());
System.out.println("訊息傳送成功!");
} catch (IOException e) {
e.printStackTrace();
}
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
```
5. 打包服務端程式碼,該步驟可忽略,僅為了測試客戶端正確性, 為了在打包時附帶其全部依賴,此處藉助Spring的打包外掛,在pom中新增以下內容:
```xml
org.springframework.boot 1.5.6.RELEASE
repackage
```
注意:Spring-boot打包外掛會自動獲取專案中的主函式,必須保證主函式只有一個,所以需要暫時註釋客戶端的主函式,最後執行maven的package,得到jar包
6. 將jar上傳至虛擬機器並執行
```shell
java -jar ZookeeperTest-1.0-SNAPSHOT.jar
```
若沒有其他問題則客戶端依然可以正常連線伺服器傳送訊息;
以上便是使用Zookeeper實現服務註冊和服務發現的具體步驟,在實際開發中,我們可能還會將提供的服務部署為叢集,這時可將叢集中的各個服務資訊作為子節點註冊到指定節點下,客戶端監聽該節點變化,獲取子節點列表從而獲取到服務列表,還可以在此基礎上加上負載均衡演算法實現對服務列表的合理訪問; 如圖:
![image-20200609103431764](https://gitee.com/yangyuanhuyyh/MDimage/raw/master/uPic/image-20200609103431