Adhesive框架中是分布式組件客戶端首先實現的是基于Json序列化+二進制協議的Memcached客戶端。在本文中會介紹其中的實現細節。
我們先來看一下項目結構：
從這個結構大致可以看出：
1）Memcached只是其中的一個具體實現，這個組件期望提供一個ClientSocket-ClientNode-ClientCluster的基礎實現，以后可以有各種客戶端基于這種結構來實現
2）對于Memcached的實現，其中把協議部分放在的Protocol文件夾中，并且根據協議為每一個請求和響應封裝類型，也就是使用面向對象的方式而不是拼數據包的方式來封裝協議
那么現在首先來介紹基礎結構。從最底部的層次開始，最底部應該是對Socket進行一個封裝，在這里我們實現了一個ClientSocket，主要完成下面功能：
1）封裝Read、Write、Connect、Reset（因為我們實現的是Socket池，所以在Socket使用之后，歸還池之前需要重置）操作
2）封裝Socket基本狀態，包括創建時間、忙碌時間、閑置時間、發生錯誤時的回調方法
?
在ClientSocket之上的一層是ClientNode，也就是一個節點的客戶端，很明顯，這里需要做的是Socket連接池，具體完成的工作有：
1）進行連接池的維護，包括移除空閑超時的Socket、強制結束忙碌時間過長的Socket、補充新的Socket到連接池的下限
2）初始化池、結束池、從池獲取Socket、把使用后的Socket返回池、創建非池Socket
在正常使用的時候，所有Socket都從池中獲取，如果整個Node不可用，那么我們定時創建非池Socket來測試Node是否恢復
?
在ClientNode之上的是ClientCluster，也就是集群，對于需要客戶端進行一致性哈希分發節點的分布式組件來說，這層就很必要了，完成的功能主要有：
1）初始化集群、使用一致性哈希從集群獲得節點、直接獲得ClientSocket
2）在節點出錯的時候進行重新節點分配、嘗試恢復出錯的節點
?
ClientCluster是使用ClientNodeLocator來分配節點的，其中的算法也就是一致性哈希算法。
之前說過節點有權重的概念，在這里也就是通過虛擬節點的數量來設置節點權重，權重越高分配到Key的數量也就會越多。
?
在ClientCluster之上還封裝了一層AbstractClient，也就是直接面向用戶的API入口。
public
abstract
class
AbstractClient<T>
where
T : AbstractClient<T>,
new
()
完成的功能有：
1）保存所有的Cluster，初始化Cluster
2）獲取具體的XXXClient的實現，比如MemcachedClient
?
很明顯，我們的第一個實現MemcachedClient是繼承了AbstractClient：
public
partial
class
MemcachedClient : AbstractClient<MemcachedClient>
在這里使用了部分類，內部的實現都放在了MemcachedClient_Internal.cs中，而對外的API都放在了MemcachedClient.cs中。
?
對于Memcached的二進制協議，我們首先是實現一個頭的格式包：
[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;
}
由于我們會直接把結構打包為字節數組，所以這里聲明了結構的內存布局。在Protocol.cs中，我們有一些實用的方法，比如結構和字節數組雙向轉換的實現：
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)));
}
在定義了頭之后，我們就可以封裝一個抽象的請求包了：
只要實現這個包，然后調用其GetBytes方法就可以直接獲得需要發送的請求數據包，它會在內部處理Header和Body數據的打包。
比如，我們來看一個Set操作的包實現：
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(
"過期時間不能超過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;
}
}
在這里，我們只是實現了抽象方法來為基類提供沒有的數據，并不需要關心數據是如何打包的。那么，之后發送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} 上執行操作 {1} 得到了不正確的回復 Key : {2} -> {3}"
,
socket.Endpoint.ToString(),
requestPackage.Opcode,
key,
responsePackage.ResponseStatus);
}
}
else
{
LocalLoggingService.Error(
"在 {0} 上執行操作 {1} 沒有得到回復 Key : {2}"
,
socket.Endpoint.ToString(),
requestPackage.Opcode,
key);
}
}
}
}
return
false
;
}
1）首先是獲取到Cluster，再獲取到池中的Socket
2）然后初始化一個SetRequestPackage，再通過GetBytes獲得數據
3）直接把數據寫入Socket
4）通過ResponsePackageCreator來獲得返回的數據包
?
很明顯，ResponsePackageCreator和AbstractRequestPackage的意圖差不多，用來把響應的數據包封裝成我們需要的數據，其中有一個：
internal
static
GeneralResponsePackage GetPackage(ClientSocket socket)
獲得的是一個通用的響應數據包：
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
;
}
}
}
}
在這里基本的信息都有了，比如操作代碼、響應狀態、Key、Value、版本號。正因為Memcached的協議比較簡單，所有的響應包都是這么一個格式，所以我們并沒有實現特殊的響應包。如果要實現的話，只需要在類頭部標記OpCode并且繼承GeneralResponsePackage，ResponsePackageCreator會自動返回相應的子類：
[AttributeUsage(AttributeTargets.Class)]
internal
class
ResponsePackageAttribute : Attribute
{
internal
Opcode Opcode { get;
private
set; }
internal
ResponsePackageAttribute(Opcode opcode)
{
this
.Opcode = opcode;
}
}
在獲得了響應之后，通過判斷ResponseStatus來知道響應是否正確，并且記錄相關日志即可。這么一來，數據一去一回以及協議如何實現的整個過程就介紹完了。下面，我們再介紹一下客戶端中幾個特色功能的實現。
?
1）獲取一組Key功能。由于一個集群會有多個節點，所以要獲取一組Key，我們首先需要把Key按照節點分類，然后對于不同的節點，采用并行的方式同時獲取，這樣速度會很快，代碼片段如下：
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的存儲，有的時候我們的Value是一個列表，那么我們可以有兩種方式完成這個功能。第一種就是直接把列表序列化作為一個Value保存，優點是簡單，缺點是如果以后需要修改的話需要整個列表取出，修改后再把整個列表保存進去，并且由于Memcached Value大小的限制，這么做也不能保存大列表；第二種方式是一個Value保存列表中的一個項，再使用一個KeyValue來保存其中每一項的ID，這么優點是修改方便，獲取的數據可以是列表中的一部分，缺點是實現麻煩，要考慮并發問題、要維護另外一個KeyValue來保存所有的ID。在這里，我們封裝了后一種方式的實現。
3）Locker功能。使用Memcached完成鎖的功能其實很簡單，我們只需要在獲取鎖的時候判斷Add一個空值是否成功，如果不成功則表示占有，等待一段時間嘗試獲取，一直到超時，在返回鎖的時候刪除這個項即可。在這里，我們封裝了MemcachedLocker來完成這個功能。
更多文章、技術交流、商務合作、聯系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
