手把手教你用netty擼一個ZkClient

時間 2019-12-20

原文原文鏈接

原文地址: https://juejin.im/post/5dd296c0e51d4508182449a6html

前言

有這個想法的原因是前一陣子突發奇想, 想嘗試能不能直接利用js鏈接到zookeeper, 從而獲取到dubbo的註冊信息.java

後來一番查找資料後, 發現因爲純js不支持tcp socket通信, 因此純js是沒法實現的. 可是發現有些大神卻使用nodeJs實現zk的客戶端. 這就成功地激起了個人興趣. 簡單地研究了一下zk通訊協議後, 我開始嘗試徒手擼一個zk的客戶端.固然是用java實現node

構思

zookeeper的通訊協議是一種典型的"header/content"結構, 在header裏面指定了content的字節數, content就是具體的報文數據.mysql

既然是header/content結構, 那麼很容易就能想到利用netty的LengthFieldPrepender來進行編碼, 以及利用LengthFieldBasedFrameDecoder來進行解碼. 有了netty這一大神器, 作什麼事情都能事半功倍.所以決定了使用netty來進行開發.git

客戶端選型決定好了以後, 還得須要有個服務端來進行調試. 從協議上看, zk不一樣的版本之間應該不會存在太多的兼容性問題, 可是差別確定是存在的. 因此爲了方便起見,咱們這裏限定了服務端的zookeeper的版本是3.4.12, 更高的版本或更低的版本沒有作過嚴格的兼容測試.github

備註: 爲了簡化工做量, 除了版本外, 該客戶端也只在單機模式測試過, 並無驗證過在集羣模式上是否能跑通.[捂臉]redis

準備工做

stat path [watch]
    set path data [version]
    ls path [watch]
    delquota [-n|-b] path
    ls2 path [watch]
    setAcl path acl
    setquota -n|-b val path
    history 
    redo cmdno
    printwatches on|off
    delete path [version]
    sync path
    listquota path
    rmr path
    get path [watch]
    create [-s] [-e] path data acl
    addauth scheme auth
    quit 
    getAcl path
    close 
    connect host:port

如上面列表所示, zkCli爲咱們提供了不少命令, 本文咱們將實現三個具備表明性的命令:sql

1. connect host:port

這個命令實際上是用來跟服務端創建會話的.
  爲了不跟socket創建tcp鏈接的connect方法相混淆, 我更願意把它稱做"login", 
  因此在實現的時候, 它對應的方法名也就是login

2. create [-s] [-e] path data acl

這個命令是用來建立zk的node節點的.
   其中包括持久性節點, 持久性順序節點, 臨時性節點, 臨時性順序節點等,
   還能夠指定ACL權限

3. ls path [watch]

ls命令就是列舉zk某個路徑下的全部子路徑,
   在具體實現裏, 我把這個命令叫作getChildren

在zookeep的通訊協議裏面, connect命令(login)是其餘全部命令的必要前置條件. 由於做爲一個客戶端, 你必須跟服務端創建了會話以後,下面的命令請求才能被服務端接受和處理.數組

而除了connect命令以外, 其餘的全部的命令其實都是截然不同的. 所以你會發現, 理解了create命令和ls命令以後, 再實現其餘命令也是很簡單的事情的, 只須要了解它們的通訊協議,其餘的都是照葫蘆畫瓢的事情了.緩存

固然瞭解它們的通訊協議並非個簡單的事情, 並且每個命令的報文結構都不大相同. 實際上在碼代碼的時候, 百分之七八十的精力基本都耗在了理解每一個命令的報文結構上面.

代碼實現

來看具體實現以前, 先來看一下項目的總結結構:

1. bean包
     封裝了每一個命令須要的字段參數, 在序列化報文時只須要序列化對應的bean便可. 一樣, 在服務端返回內容時, 也只須要把報文序列化成對應的對象便可.
  2. factories包
      上面提到過, zk的每一個命令的報文結構都是不同的,因此在序列化和反序列化時, 對應到netty的codec也是不同的.這個實現了一個codec靜態方法工廠, 須要的時候直接從codec工廠拿對應的codec便可.
  3. registrys包
     其實就是一個緩存中心, 緩存了每一個命令對應的requestId和codec, 在服務端返回時, 從這個緩存中心根據requestId拿到對應codec來進行反序列化
  4. utils包
     一些工具類, 不須要多解釋
  
  5. zkcodec包
     每一個命令對應的codec和handler實現
  6. NettyZkClient類
     就是本文要介紹的zk客戶端了
  7. test    
     爲了方便調試準備的單元測試, 先了解代碼實現原理的可用直接從這個單元測試入手

看完代碼結構後, 咱們再來看每一個命令的具體實現.

login命令

首先來看一下login命令的通訊報文結構體, 以下圖所示:

簡單介紹一下每一個字段的含義, 具體的含義你們能夠在網上搜索作更深刻的瞭解:

1. header
    上面提到, zk的每一個報文都是header/content模式, 其中header佔用4個字節, 表示接下來的content的長度(字節數)
 2. protocolVersion
    顧名思義, 這個字段表示協議的版本號,用4個字節表示. 這裏咱們寫死爲0便可(好浪費~~~)
 3. lastZxidSeen
    等下咱們會看到, zk服務端每次的響應都會返回一個zxid.顧名思義, 這個結構就是表示客戶端接收到最新的zxid.用8個字節表示. 因爲login通常都是第一次跟服務端通信, 因此這裏也是寫死爲0便可
 4. timeout 
    login請求的目的是爲了跟zk服務端創建會話, 這個參數表示的是會話的有效時間, 用四個字節表示
 5. senssionId
    會話ID, 用8個字節表示, 用於咱們尚未創建會話,因此這個也是直接寫死爲0便可
 6. passwordLen
    用4個字節來表示接下來的密碼的字節數
 7. password
    passwordLen個字節的密碼, 用bytes[]表示
 8. readOnly
    boolean類型的,因此用一個字節表示便可

知道了報文結構以後, 咱們就能夠開始寫代碼了.
首先定義一個ZkLoginRequest的bean.在java裏面, 8個字節能夠用long類型表示, 4個字節能夠用int表示, String類型能夠很簡單地轉換成byte數組. 因此最後定義的bean以下所示:

public class ZkLoginRequest implements Serializable {
  private Integer protocolVersion;
  private Long lastZxidSeen;
  private int timeout;
  private Long sessionId;
  private String password;
  private boolean readOnly = true;
    
}

由於zk的通信是基於字節的, 因此咱們僅僅定義java對象是不行的, 還須要把java對象轉換成字節, 才能發送服務端.並且服務端只會接收header/content形式的報文, 因此咱們還得計算出整個java對象序列化以後的字節數, 把它賦值到header中去.

幸運的是, 這兩個工做netty都爲咱們提供了很好的工具, 咱們直接使用就能夠了.

實現ZkLoginCodec把java對象轉換成ByteBuf

ZkLoginCodec包括encode和decode兩部分, 其中decode是用於解碼服務端的響應的, encode是用於編碼發送請求的, 以下所示, 把ZkLoginRequest轉換成netty的ByteBuf

@Override
 protected void encode(ChannelHandlerContext ctx, ZkLoginRequest msg, ByteBuf outByteBuf) throws Exception {
     outByteBuf.writeInt(msg.getProtocolVersion());
     outByteBuf.writeLong(msg.getLastZxidSeen());
     outByteBuf.writeInt(msg.getTimeout());
     outByteBuf.writeLong(msg.getSessionId());
     String passWord = msg.getPassword();
     SerializeUtils.writeStringToBuffer(passWord, outByteBuf);
     outByteBuf.writeBoolean(msg.isReadOnly());
 }

直接使用netty內置的LengthFieldPrepender

netty內置的LengthFieldPrepender能夠把報文轉換成header/content形式的結構, 其中構造參數表示header所佔的字節數, 咱們這裏是4個字節, 因此是4.

// 編碼器, 給報文加上一個4個字節大小的HEADER
   nioSocketChannel.pipeline().addLast(new LengthFieldPrepender(4))

ZkLoginRequest對象通過這兩個codec編碼以後,zk服務端就能正確解析它的報文了, 如無心外的話, 服務端會針對咱們的這個socket創建會話, 而後給咱們響應一個報文, 報文中會包含sessionId. 響應的結構體以下所示:

能夠看到, 響應報文跟咱們的請求報文是差很少的,除了sessionId, 其餘的基本是原封不動地給咱們返回來了. 因此咱們這裏對響應報文含義很少作解釋. 直接來看看如何解析服務端返回的響應報文.

從上圖咱們能夠看到, 返回的報文也是header/content形式, 因此咱們仍是可使用netty內置的解碼器來獲取header和content.

不熟悉LengthFieldBasedFrameDecoder的同窗能夠先看看netty的官網, 這裏不對這個類的參數作過多解釋, 只須要知道咱們是想跳過header,只獲取響應報文的content部分便可

nioSocketChannel.pipeline()
                        // 解碼器, 將HEADER-CONTENT格式的報文解析成只包含CONTENT
        .addLast(ZkLoginHandler.LOGIN_LENGTH_FIELD_BASED_FRAME_DECODER,new LengthFieldBasedFrameDecoder(2048, 0, 4, 0, 4))

ZkLoginCodec把content反序列成ZkLoginResp

也就是ZkLoginCodec的decode部分

@Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        ZkLoginResp zkLoginResp = new ZkLoginResp();
        zkLoginResp.setProtocolVersion(in.readInt());
        zkLoginResp.setTimeout(in.readInt());
        zkLoginResp.setSessionId(in.readLong());
        String password = SerializeUtils.readStringToBuffer(in);
        zkLoginResp.setPassword(password);
        zkLoginResp.setReadOnly(in.readBoolean());
        out.add(zkLoginResp);
    }

完成這一步後, 咱們的客戶端就成功地跟服務端創建了會話了, 後面就能夠愉快地發送其餘請求了

create命令

opCode和requestHeader

在開始實現create命令以前, 先來了解一下兩個術語,這兩個術語不單是create命令須要用到的, 等下要實現的getChildren命令也一樣須要用到.

opCode

zk的服務端跟客戶端約定好了, 每個命令都對應一個opCode, 客戶端發送命令請求時必須帶上這個opCode, 服務端才能知道如何去處理這個命令. 例如create命令對應的opCode是1, 等下要實現的getChildren命令的opCode是8

requestHeader

這個header不要跟header/content中的header混淆了. requestHeader仍是content的一部分, 它包含了兩個字段, 每一個字段佔4個字節, 以下圖:

1. xid, 通俗點理解的就是requestId, 客戶端維護這個xid的惟一性, 服務端返回響應時會原封不動的返回這個xid,
   這樣客戶端就能夠知道服務端返回的報文時對應哪一個請求的了.畢竟socket通信都是異步的.
   
   2. type
      這個更好理解, 就是上面的opcode

create命令的報文結構

說真的, create命令的報文有一丁點複雜, 因此
在看createRequest的報文結構以前,還得先了解另一個概念: ACL權限;

zookeeper的ACL權限涉及到如下3個點:

scheme 身份的認證有4種方式：
id 受權的對象
permission 權限

具體能夠看這篇博客, 說得比較清楚.

在本文中咱們寫死了scheme是"world", id是"anyone", permission是31(轉成二進制即11111,擁有CREATE、READ、WRITE、DELETE、ADMIN五種權限)

zk中的ACL的報文結構以下所示:

1. perms是permission的簡寫, 4個字節表示
  2. 由於scheme是用字符串表示的, 因此須要用4個字節表示scheme字符串的長度, 用schemelen個字節表示scheme
  3. id也是用字符串表示的, 跟scheme同理

瞭解了requestHeader和ACL的結構體以後, createRequest的報文結構也就比較好理解了, 以下圖所示:

1. requestHeader
   包含了xid和type
2. pathLen
   要建立的path的字符串長度
3. path
   要建立的path, 例如你想在zk上建立一個/dubbo節點, path就是"/path"
4. dataLen
   要建立的path下的data的長度
5. data
   要建立的path下的數據, 例如"192.168.99.101:2181"
6. aclListSize
   zk的ACL權限是list形式的,表示不一樣的權限控制維度
7. aclList
   aclListSize個ACL結構體
8. flags
   該節點的類型, 能夠是持久性節點, 持久性順序節點, 臨時節點, 臨時順序節點4種

接下來的工做就是login差很少了:

實現一個ZkCreateCodec把ZkCreateRequest轉化成ByteBuf
利用LengthFieldPrepender把ByteBuf構建成Header/content的報文結構
利用LengthFieldBasedFrameDecoder從服務端的響應解析出content的內容
把content內容的ByteBuf轉換成ZkCreateResponse

其中createRes的報文結構以下圖所示:

1. xid
   這個就是createReq中requestHeader的xid
2. zxid
   這個能夠跟login報文中的lastZxidSeen關聯起來, 能夠理解爲服務端的xid
3. errcode
   errcode爲0表示正常響應, 若是不爲0,說明有異常出現, 後面的path字段也會爲空
4. pathLen和path
   其實也是createReuest中的path和pathLen

3. getChildren命令(ls path)

若是上面的create命令能理解的話, 那getChildren命令也就很容易理解了, 二者只有報文結構不同,於是codec也會有一點點不一樣.

getChildrenRequest的報文結構:

響應的結構體:

運行代碼

要想快速地體驗一下這個簡陋的zkClient的話, 能夠直接從單元測試入手:

public class ZkClientTest {
    @Test
    public void testZkClient() throws Exception {
        // NettyZkClient的構造方法裏面會調用login() 跟服務端創建會話
        NettyZkClient nettyZkClient = new NettyZkClient(30000);

        // 建立一個臨時順序節點
        ZkCreateResponse createResponse = nettyZkClient.create("/as", 12312, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(new Gson().toJson(createResponse));

        // 獲取/下的全部子路徑
        List<String> list =  nettyZkClient.getChildren("/");
        System.out.println(new Gson().toJson(list));

    }
}