Kafka與.net core（二）zookeeper

1.zookeeper簡單介紹

1.1作用

zookeeper的作用是存儲kafka的服務器信息，topic信息，和cunsumer信息。如下圖：

而zookeeper是個什么東西呢？簡單來說就是一個具有通知機制的文件系統，引用網路上的一張圖

可以看出來zookeeper是一個樹形的文件結構，我們可以自定義node與node的值，並對node進行監視，當node的結構或者值變化時，我們可以收到通知。

1.2node類型

1）PERSISTENT-持久化目錄節點

客戶端與zookeeper斷開連接后，該節點依舊存在

2）PERSISTENT_SEQUENTIAL-持久化順序編號目錄節點

客戶端與zookeeper斷開連接后，該節點依舊存在，只是Zookeeper給該節點名稱進行順序編號

3）EPHEMERAL-臨時目錄節點

客戶端與zookeeper斷開連接后，該節點被刪除

4）EPHEMERAL_SEQUENTIAL-臨時順序編號目錄節點

客戶端與zookeeper斷開連接后，該節點被刪除，只是Zookeeper給該節點名稱進行順序編號

2.zookeeper命令操作

連接zookeeper

[root@iz2zei2y693gtrgwlibzlwz ~]# zkCli.sh -server ip:2181

查看zookeeper的所有節點

ls /

查看某個節點的子節點

ls /brokers

創建節點

 create /testaa dataaaa

獲取節點的值

get /testaa

設置節點值

 set /testaa aaabbb

刪除節點

 delete /testaa

這么看來實際上zookeeper跟數據庫類似也是CURD操作，我們再來看看zookeeper的安全控制ACL

3.zookeeper的ACL

3.1ZK的節點有5種操作權限：

CREATE、READ、WRITE、DELETE、ADMIN 也就是 增、刪、改、查、管理權限，這5種權限簡寫為crwda(即：每個單詞的首字符縮寫)

3.2身份的認證有4種方式：

world：默認方式，相當於全世界都能訪問

auth：代表已經認證通過的用戶(cli中可以通過addauth digest user:pwd 來添加當前上下文中的授權用戶)

digest：即用戶名:密碼這種方式認證，這也是業務系統中最常用的

ip：使用Ip地址認證

3.3ACL實例

默認創建的時world方式，任何人都能訪問，我們新建一個測試node節點

[zk: 3:2181(CONNECTED) 11] create /cys cys

訪問一一下

[zk: 3:2181(CONNECTED) 13] get /cys

查看一下Acl

[zk: 3:2181(CONNECTED) 15] getAcl /cys

下面我們設置一下他的用戶

 命令為：

1）增加一個認證用戶

addauth digest 用戶名:密碼明文

eg. addauth digest user1:password1

2）設置權限

setAcl /path auth:用戶名:密碼明文:權限

eg. setAcl /test auth:user1:password1:cdrwa

3）查看Acl設置

getAcl /path

 具體操作如下：

addauth digest cys:123456
setAcl /cys auth:cys:123456:crwda

我們ctrl+c退出zkCli，重新連接一下，然后查詢

get /cys

結果如下：

提示我們認證失敗，我們登陸一下

addauth digest cys:123456

結果如下：

我們在查看一下/cys節點的Acl

 

可以看出來用戶cys對應的密碼（加密后的）和權限cdrwa

4..net core 操作

4.1新建server項目，引入ZookeeperNetEx這個nuget包

Server端代碼

using org.apache.zookeeper;
using org.apache.zookeeper.data;

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace ConsoleApp3
{
    class Program
    {
        static void Main(string[] args)
        {
            while (true)
            {
                Console.WriteLine("輸入path");
                var path = Console.ReadLine();
                string address = "39.**.**.**:2181";
                ZooKeeper _zooKeeper = new ZooKeeper(address, 1000 * 1000, null);
                ZooKeeper.States states = _zooKeeper.getState();
                //是否存在
                Task<org.apache.zookeeper.data.Stat> stat = _zooKeeper.existsAsync(path);
                stat.Wait();
                if (stat.Result != null && stat.Status.ToString().ToLower() == "RanToCompletion".ToLower())
                {
                    //已存在
                    Console.WriteLine($"{path}已存在");
                }
                else
                {
                    Console.WriteLine("輸入data");
                    var data = Console.ReadLine();
                    //創建
                    Task<string> task = _zooKeeper.createAsync(path, System.Text.Encoding.UTF8.GetBytes(data), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                    task.Wait();
                    if (!string.IsNullOrEmpty(task.Result) && task.Status.ToString().ToLower() == "RanToCompletion".ToLower())
                    {
                        Console.WriteLine($"{path}創建成功");
                    }
                }
                Console.WriteLine("輸入set data");
                var dataA = Console.ReadLine();
                //set值
                Task<org.apache.zookeeper.data.Stat> statA = _zooKeeper.setDataAsync(path, System.Text.Encoding.UTF8.GetBytes(dataA));
                statA.Wait();
                if (statA.Result != null && statA.Status.ToString().ToLower() == "RanToCompletion".ToLower())
                {
                    Console.WriteLine("set 成功");
                }

                Console.WriteLine("輸入子path");
                var childpath = Console.ReadLine();
     
                Console.WriteLine("輸入子data");
                var childdata = Console.ReadLine();
                Task<string> childtask = _zooKeeper.createAsync(childpath, System.Text.Encoding.UTF8.GetBytes(childdata), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                childtask.Wait();
                if (!string.IsNullOrEmpty(childtask.Result) && childtask.Status.ToString().ToLower() == "RanToCompletion".ToLower())
                {
                    Console.WriteLine($"{childpath}創建成功");
                }

                Console.ReadLine();
                _zooKeeper.closeAsync().Wait();
            }
            ////刪除
            //Console.WriteLine("輸入delete path");
            //var pathB = Console.ReadLine();
            //Task taskA = _zooKeeper.deleteAsync(pathB);
            //taskA.Wait();
            //if (taskA.Status.ToString().ToLower() == "RanToCompletion".ToLower())
            //{
            //    Console.WriteLine("delete 成功");
            //}

            ////獲取數據
            //Task<DataResult> dataResult = _zooKeeper.getDataAsync(path, new NodeWatcher());
            //dataResult.Wait();
            //if (dataResult.Result != null && dataResult.Status.ToString().ToLower() == "RanToCompletion".ToLower())
            //{
            //    Console.WriteLine(Encoding.UTF8.GetString(dataResult.Result.Data));
            //}
        }
    }

}

4.2新建client項目，引入ZookeeperNetEx這個nuget包

客戶端代碼

using org.apache.zookeeper;
using org.apache.zookeeper.data;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace Client
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("輸入path");
            var path = Console.ReadLine();
            string address = "39.**.**.**:2181";
            ZooKeeper _zooKeeper = new ZooKeeper(address, 10 * 1000, new DefaultWatcher());
　　　　　　　//用戶登陸
            _zooKeeper.addAuthInfo("digest", System.Text.Encoding.Default.GetBytes("cys:123456"));
            //獲取child
            var getresult = _zooKeeper.getChildrenAsync(path, true);
            getresult.Wait();

            //獲取數據
            Task<DataResult> dataResult = _zooKeeper.getDataAsync(path,true);
            dataResult.Wait();
            Thread.Sleep(100000);
        }
    }

    public class DefaultWatcher : Watcher
    {
        internal static readonly Task CompletedTask = Task.FromResult(1);

        /// <summary>
        /// 接收通知
        /// </summary>
        /// <param name="event"></param>
        /// <returns></returns>
        public override Task process(WatchedEvent @event)
        {
            Console.WriteLine(string.Format("接收到ZooKeeper服務端的通知，State是：{0}，EventType是：{1}，Path是：{2}", @event.getState(), @event.get_Type(), @event.getPath() ?? string.Empty));
            return CompletedTask;
        }
    }
}

這樣當server端操作的時候，client端會通過watcher收到通知