Adhesive框架中是分布式組件客戶端首先實(shí)現(xiàn)的是基于Json序列化+二進(jìn)制協(xié)議的Memcached客戶端。在本文中會介紹其中的實(shí)現(xiàn)細(xì)節(jié)。

我們先來看一下項(xiàng)目結(jié)構(gòu)：

從這個(gè)結(jié)構(gòu)大致可以看出：

1）Memcached只是其中的一個(gè)具體實(shí)現(xiàn)，這個(gè)組件期望提供一個(gè)ClientSocket-ClientNode-ClientCluster的基礎(chǔ)實(shí)現(xiàn)，以后可以有各種客戶端基于這種結(jié)構(gòu)來實(shí)現(xiàn)

2）對于Memcached的實(shí)現(xiàn)，其中把協(xié)議部分放在的Protocol文件夾中，并且根據(jù)協(xié)議為每一個(gè)請求和響應(yīng)封裝類型，也就是使用面向?qū)ο蟮姆绞蕉皇瞧磾?shù)據(jù)包的方式來封裝協(xié)議

那么現(xiàn)在首先來介紹基礎(chǔ)結(jié)構(gòu)。從最底部的層次開始，最底部應(yīng)該是對Socket進(jìn)行一個(gè)封裝，在這里我們實(shí)現(xiàn)了一個(gè)ClientSocket，主要完成下面功能：

1）封裝Read、Write、Connect、Reset（因?yàn)槲覀儗?shí)現(xiàn)的是Socket池，所以在Socket使用之后，歸還池之前需要重置）操作

2）封裝Socket基本狀態(tài)，包括創(chuàng)建時(shí)間、忙碌時(shí)間、閑置時(shí)間、發(fā)生錯(cuò)誤時(shí)的回調(diào)方法

?

在ClientSocket之上的一層是ClientNode，也就是一個(gè)節(jié)點(diǎn)的客戶端，很明顯，這里需要做的是Socket連接池，具體完成的工作有：

1）進(jìn)行連接池的維護(hù)，包括移除空閑超時(shí)的Socket、強(qiáng)制結(jié)束忙碌時(shí)間過長的Socket、補(bǔ)充新的Socket到連接池的下限

2）初始化池、結(jié)束池、從池獲取Socket、把使用后的Socket返回池、創(chuàng)建非池Socket

在正常使用的時(shí)候，所有Socket都從池中獲取，如果整個(gè)Node不可用，那么我們定時(shí)創(chuàng)建非池Socket來測試Node是否恢復(fù)

?

在ClientNode之上的是ClientCluster，也就是集群，對于需要客戶端進(jìn)行一致性哈希分發(fā)節(jié)點(diǎn)的分布式組件來說，這層就很必要了，完成的功能主要有：

1）初始化集群、使用一致性哈希從集群獲得節(jié)點(diǎn)、直接獲得ClientSocket

2）在節(jié)點(diǎn)出錯(cuò)的時(shí)候進(jìn)行重新節(jié)點(diǎn)分配、嘗試恢復(fù)出錯(cuò)的節(jié)點(diǎn)

?

ClientCluster是使用ClientNodeLocator來分配節(jié)點(diǎn)的，其中的算法也就是一致性哈希算法。

之前說過節(jié)點(diǎn)有權(quán)重的概念，在這里也就是通過虛擬節(jié)點(diǎn)的數(shù)量來設(shè)置節(jié)點(diǎn)權(quán)重，權(quán)重越高分配到Key的數(shù)量也就會越多。

?

在ClientCluster之上還封裝了一層AbstractClient，也就是直接面向用戶的API入口。

    
      public
    
    
      abstract
    
    
      class
    
     AbstractClient<T> 
    
      where
    
     T : AbstractClient<T>, 
    
      new
    
    ()
  

完成的功能有：

1）保存所有的Cluster，初始化Cluster

2）獲取具體的XXXClient的實(shí)現(xiàn)，比如MemcachedClient

?

很明顯，我們的第一個(gè)實(shí)現(xiàn)MemcachedClient是繼承了AbstractClient：

    
      public
    
    
      partial
    
    
      class
    
     MemcachedClient : AbstractClient<MemcachedClient>
  

在這里使用了部分類，內(nèi)部的實(shí)現(xiàn)都放在了MemcachedClient_Internal.cs中，而對外的API都放在了MemcachedClient.cs中。

?

對于Memcached的二進(jìn)制協(xié)議，我們首先是實(shí)現(xiàn)一個(gè)頭的格式包：

        [StructLayout(LayoutKind.Sequential, Pack = 1)]

    
    
      internal
    
    
      struct
    
     Header

    {

        
    
      internal
    
    
      byte
    
     Magic;



        
    
      internal
    
    
      byte
    
     Opcode;



        
    
      internal
    
    
      ushort
    
     KeyLength;



        
    
      internal
    
    
      byte
    
     ExtraLength;



        
    
      internal
    
    
      byte
    
     DataType;



        
    
      internal
    
    
      ushort
    
     Reserved;



        
    
      internal
    
    
      uint
    
     TotalBodyLength;



        
    
      internal
    
    
      uint
    
     Opaque;



        
    
      internal
    
    
      ulong
    
     Version;

    }
  

由于我們會直接把結(jié)構(gòu)打包為字節(jié)數(shù)組，所以這里聲明了結(jié)構(gòu)的內(nèi)存布局。在Protocol.cs中，我們有一些實(shí)用的方法，比如結(jié)構(gòu)和字節(jié)數(shù)組雙向轉(zhuǎn)換的實(shí)現(xiàn)：

    
      internal
    
    
      static
    
     T BytesToStruct<T>(
    
      this
    
    
      byte
    
    [] rawData)

        {

            T result = 
    
      default
    
    (T);

            RespectEndianness(
    
      typeof
    
    (T), rawData);

            GCHandle handle = GCHandle.Alloc(rawData, GCHandleType.Pinned);

            
    
      try
    
    

            {

                IntPtr rawDataPtr = handle.AddrOfPinnedObject();

                result = (T)Marshal.PtrToStructure(rawDataPtr, 
    
      typeof
    
    (T));

            }

            
    
      finally
    
    

            {

                handle.Free();

            }

            
    
      return
    
     result;

        }



        
    
      internal
    
    
      static
    
    
      byte
    
    [] StructToBytes<T>(
    
      this
    
     T data)

        {

            
    
      byte
    
    [] rawData = 
    
      new
    
    
      byte
    
    [Marshal.SizeOf(data)];

            GCHandle handle = GCHandle.Alloc(rawData, GCHandleType.Pinned);

            
    
      try
    
    

            {

                IntPtr rawDataPtr = handle.AddrOfPinnedObject();

                Marshal.StructureToPtr(data, rawDataPtr, 
    
      false
    
    );

            }

            
    
      finally
    
    

            {

                handle.Free();

            }

            RespectEndianness(
    
      typeof
    
    (T), rawData);

            
    
      return
    
     rawData;

        }



        
    
      private
    
    
      static
    
    
      void
    
     RespectEndianness(Type type, 
    
      byte
    
    [] data)

        {

            var fields = type.GetFields(BindingFlags.NonPublic | BindingFlags.Instance).Select(field => 
    
      new
    
    

            {

                Field = field,

                Offset = Marshal.OffsetOf(type, field.Name).ToInt32()

            }).ToList();



            fields.ForEach(item => Array.Reverse(data, item.Offset, Marshal.SizeOf(item.Field.FieldType)));

        }
  

在定義了頭之后，我們就可以封裝一個(gè)抽象的請求包了：

只要實(shí)現(xiàn)這個(gè)包，然后調(diào)用其GetBytes方法就可以直接獲得需要發(fā)送的請求數(shù)據(jù)包，它會在內(nèi)部處理Header和Body數(shù)據(jù)的打包。

比如，我們來看一個(gè)Set操作的包實(shí)現(xiàn)：

    
      internal
    
    
      class
    
     SetRequestPackage : AbstractRequestPackage

    {

        
    
      private
    
     TimeSpan expireSpan;

        
    
      private
    
    
      byte
    
    [] valueBytes;

        
    
      private
    
    
      ulong
    
     version;



        
    
      public
    
    
      override
    
     Opcode Opcode

        {

            get { 
    
      return
    
     Opcode.Set; }

        }



        
    
      internal
    
     SetRequestPackage(
    
      string
    
     key, 
    
      byte
    
    [] valueBytes, TimeSpan expireSpan, 
    
      ulong
    
     version)

            : 
    
      base
    
    (key)

        {

            
    
      if
    
     (expireSpan > TimeSpan.FromDays(30))

                
    
      throw
    
    
      new
    
     ArgumentOutOfRangeException(
    
      "過期時(shí)間不能超過30天!"
    
    );

            
    
      this
    
    .expireSpan = expireSpan;

            
    
      this
    
    .valueBytes = valueBytes;

            
    
      this
    
    .version = version;

        }



        
    
      internal
    
     SetRequestPackage(
    
      string
    
     key, 
    
      string
    
    
      value
    
    , TimeSpan expireSpan, 
    
      ulong
    
     version)

            : 
    
      this
    
    (key, Encoding.UTF8.GetBytes(
    
      value
    
    ), expireSpan, version)

        {

        }



        
    
      internal
    
     SetRequestPackage(
    
      string
    
     key, 
    
      string
    
    
      value
    
    , 
    
      ulong
    
     version)

            : 
    
      this
    
    (key, Encoding.UTF8.GetBytes(
    
      value
    
    ), TimeSpan.FromDays(30), version)

        {

        }



        
    
      internal
    
     SetRequestPackage(
    
      string
    
     key, 
    
      byte
    
    [] valueBytes, 
    
      ulong
    
     version)

            : 
    
      this
    
    (key, valueBytes, TimeSpan.FromDays(30), version)

        {

        }



        
    
      protected
    
    
      override
    
    
      ulong
    
     GetVersion()

        {

            
    
      return
    
     version;

        }



        
    
      protected
    
    
      override
    
    
      byte
    
    [] GetExtraBytes()

        {

            var extraBytes = 
    
      new
    
     List<
    
      byte
    
    >();

            
    
      uint
    
     flag = 0xdeadbeef;

            extraBytes.AddRange(flag.GetBigEndianBytes());

            
    
      uint
    
     expire = Convert.ToUInt32(expireSpan.TotalSeconds);

            extraBytes.AddRange(expire.GetBigEndianBytes());

            
    
      return
    
     extraBytes.ToArray();

        }



        
    
      protected
    
    
      override
    
    
      byte
    
    [] GetValueBytes()

        {

            
    
      return
    
     valueBytes;

        }

    }
  

在這里，我們只是實(shí)現(xiàn)了抽象方法來為基類提供沒有的數(shù)據(jù)，并不需要關(guān)心數(shù)據(jù)是如何打包的。那么，之后發(fā)送Set請求的操作就很簡單了：

    
      private
    
    
      bool
    
     InternalSet(
    
      string
    
     key, 
    
      string
    
    
      value
    
    , TimeSpan expire, 
    
      ulong
    
     version)

        {

            
    
      using
    
     (var socket = GetCluster().AcquireSocket(key))

            {

                
    
      if
    
     (socket != 
    
      null
    
    )

                {

                    AbstractRequestPackage requestPackage = expire == TimeSpan.MaxValue ? 
    
      new
    
     SetRequestPackage(key, 
    
      value
    
    , version)

                            : 
    
      new
    
     SetRequestPackage(key, 
    
      value
    
    , expire, version);

                    var requestData = requestPackage.GetBytes();

                    
    
      if
    
     (requestData != 
    
      null
    
    )

                    {

                        socket.Write(requestData);

                        var responsePackage = ResponsePackageCreator.GetPackage(socket);

                        
    
      if
    
     (responsePackage != 
    
      null
    
    )

                        {

                            
    
      if
    
     (responsePackage.ResponseStatus == ResponseStatus.NoError)

                            {

                                
    
      return
    
    
      true
    
    ;

                            }

                            
    
      else
    
    
      if
    
     (responsePackage.ResponseStatus != ResponseStatus.KeyExists

                                    && responsePackage.ResponseStatus != ResponseStatus.KeyNotFound)

                            {

                                LocalLoggingService.Warning(
    
      "在 {0} 上執(zhí)行操作 {1} 得到了不正確的回復(fù) Key : {2} -> {3}"
    
    ,

                                            socket.Endpoint.ToString(),

                                            requestPackage.Opcode,

                                            key,

                                            responsePackage.ResponseStatus);

                            }

                        }

                        
    
      else
    
    

                        {

                            LocalLoggingService.Error(
    
      "在 {0} 上執(zhí)行操作 {1} 沒有得到回復(fù) Key : {2}"
    
    ,

                                       socket.Endpoint.ToString(),

                                       requestPackage.Opcode,

                                       key);

                        }

                    }

                }

            }

            
    
      return
    
    
      false
    
    ;

        }
  

1）首先是獲取到Cluster，再獲取到池中的Socket

2）然后初始化一個(gè)SetRequestPackage，再通過GetBytes獲得數(shù)據(jù)

3）直接把數(shù)據(jù)寫入Socket

4）通過ResponsePackageCreator來獲得返回的數(shù)據(jù)包

?

很明顯，ResponsePackageCreator和AbstractRequestPackage的意圖差不多，用來把響應(yīng)的數(shù)據(jù)包封裝成我們需要的數(shù)據(jù)，其中有一個(gè)：

    
      internal
    
    
      static
    
     GeneralResponsePackage GetPackage(ClientSocket socket)
  

獲得的是一個(gè)通用的響應(yīng)數(shù)據(jù)包：

    
      internal
    
    
      class
    
     GeneralResponsePackage

    {

        
    
      internal
    
     Opcode Opcode { get; set; }



        
    
      internal
    
     ResponseStatus ResponseStatus { get; set; }



        
    
      internal
    
    
      string
    
     Key { get; set; }



        
    
      internal
    
    
      byte
    
    [] ValueBytes { get; set; }



        
    
      internal
    
    
      ulong
    
     Version { get; set; }



        
    
      internal
    
    
      string
    
     Value

        {

            get

            {

                
    
      if
    
     (ValueBytes != 
    
      null
    
    )

                {

                    
    
      return
    
     Encoding.UTF8.GetString(ValueBytes);

                }

                
    
      else
    
    

                {

                    
    
      return
    
    
      null
    
    ;

                }

            }

        }

    }
  

在這里基本的信息都有了，比如操作代碼、響應(yīng)狀態(tài)、Key、Value、版本號。正因?yàn)镸emcached的協(xié)議比較簡單，所有的響應(yīng)包都是這么一個(gè)格式，所以我們并沒有實(shí)現(xiàn)特殊的響應(yīng)包。如果要實(shí)現(xiàn)的話，只需要在類頭部標(biāo)記OpCode并且繼承GeneralResponsePackage，ResponsePackageCreator會自動返回相應(yīng)的子類：

        [AttributeUsage(AttributeTargets.Class)]

    
    
      internal
    
    
      class
    
     ResponsePackageAttribute : Attribute

    {

        
    
      internal
    
     Opcode Opcode { get; 
    
      private
    
     set; }



        
    
      internal
    
     ResponsePackageAttribute(Opcode opcode)

        {

            
    
      this
    
    .Opcode = opcode;

        }

    }
  

在獲得了響應(yīng)之后，通過判斷ResponseStatus來知道響應(yīng)是否正確，并且記錄相關(guān)日志即可。這么一來，數(shù)據(jù)一去一回以及協(xié)議如何實(shí)現(xiàn)的整個(gè)過程就介紹完了。下面，我們再介紹一下客戶端中幾個(gè)特色功能的實(shí)現(xiàn)。

?

1）獲取一組Key功能。由于一個(gè)集群會有多個(gè)節(jié)點(diǎn)，所以要獲取一組Key，我們首先需要把Key按照節(jié)點(diǎn)分類，然后對于不同的節(jié)點(diǎn)，采用并行的方式同時(shí)獲取，這樣速度會很快，代碼片段如下：

                var nodeCache = 
    
      new
    
     Dictionary<ClientNode, List<
    
      string
    
    >>();

            
    
      foreach
    
     (var key 
    
      in
    
     keys)

            {

                var node = GetCluster().AcquireNode(key);

                
    
      if
    
     (!nodeCache.ContainsKey(node))

                    nodeCache.Add(node, 
    
      new
    
     List<
    
      string
    
    > { key });

                
    
      else
    
    
      if
    
     (!nodeCache[node].Contains(key))

                    nodeCache[node].Add(key);

            }



            var data = 
    
      new
    
     Dictionary<
    
      string
    
    , 
    
      string
    
    >();

            Parallel.ForEach(nodeCache, node =>
  

2）List功能。Memcached只提供了Key、Value的存儲，有的時(shí)候我們的Value是一個(gè)列表，那么我們可以有兩種方式完成這個(gè)功能。第一種就是直接把列表序列化作為一個(gè)Value保存，優(yōu)點(diǎn)是簡單，缺點(diǎn)是如果以后需要修改的話需要整個(gè)列表取出，修改后再把整個(gè)列表保存進(jìn)去，并且由于Memcached Value大小的限制，這么做也不能保存大列表；第二種方式是一個(gè)Value保存列表中的一個(gè)項(xiàng)，再使用一個(gè)KeyValue來保存其中每一項(xiàng)的ID，這么優(yōu)點(diǎn)是修改方便，獲取的數(shù)據(jù)可以是列表中的一部分，缺點(diǎn)是實(shí)現(xiàn)麻煩，要考慮并發(fā)問題、要維護(hù)另外一個(gè)KeyValue來保存所有的ID。在這里，我們封裝了后一種方式的實(shí)現(xiàn)。

3）Locker功能。使用Memcached完成鎖的功能其實(shí)很簡單，我們只需要在獲取鎖的時(shí)候判斷Add一個(gè)空值是否成功，如果不成功則表示占有，等待一段時(shí)間嘗試獲取，一直到超時(shí)，在返回鎖的時(shí)候刪除這個(gè)項(xiàng)即可。在這里，我們封裝了MemcachedLocker來完成這個(gè)功能。

Adhesive框架系列文章--分布式組件客戶端模塊實(shí)現(xiàn)