前言
介紹
[NetMQ](https://github.com/zeromq/netmq.git)是ZeroMQ的C#移植版本,它是對標准socket接口的擴展。它提供了一種異步消息隊列,多消息模式,消息過濾（訂閱）,對多種傳輸協議的無縫訪問。
當前有2個版本正在維護，版本3最新版為3.3.4，版本4最新版本為4.0.0-rc5。本文檔是對4.0.0-rc5分支代碼進行分析。
目的
對NetMQ的源碼進行學習並分析理解,因此寫下該系列文章,本系列文章暫定編寫計划如下：
- 消息隊列NetMQ 原理分析1-Context和ZObject
- 消息隊列NetMQ 原理分析2-IO線程和完成端口
- 消息隊列NetMQ 原理分析3-命令產生/處理和回收線程
- 消息隊列NetMQ 原理分析4-Session和Pipe
- 消息隊列NetMQ 原理分析5-Engine
- 消息隊列NetMQ 原理分析6-TCP和Inpoc實現
- 消息隊列NetMQ 原理分析7-Device
- 消息隊列NetMQ 原理分析8-不同類型的Socket
- 消息隊列NetMQ 原理分析9-實戰
友情提示: 看本系列文章時最好獲取源碼,更有助於理解。
IO線程
NetMQ 4.0.0底層使用的是IOCP(即完成端口)模式進行通信的(3.3.4使用的是select模型),通過異步IO綁定到完成端口,來最大限度的提高性能。這里不對同步/異步socket進行詳細介紹。稍微解釋下完成端口,為了解決每個socket客戶端使用一個線程進行通信的性能問題，完成端口它充分利用內核對象的調度，只使用少量的幾個線程來處理和客戶端的所有通信，消除了無謂的線程上下文切換，最大限度的提高了網絡通信的性能。
想詳細了解完成端口的請看完成端口(Completion Port)詳解 ,講解的比較詳細,同時對各種網絡編程模型做了簡單的介紹。
因此NetMQ通過幾個(默認1個)IO線程處理通信,上一片文章介紹了ZObejct對象,在該對象中存在許多命令的處理,實際對命令的發送,分配都是IO線程的工作。
初始化IO線程
IO線程初始化時會初始化Proactor和IOThreadMailbox
var name = "iothread-" + threadId;
m_proactor = new Proactor(name);
m_mailbox = new IOThreadMailbox(name, m_proactor, this);
Proactor對象就是用來綁定或處理完成端口用的,后面再做作詳細介紹。
IOThreadMailbox是IO線程處理的信箱,每當有命令需要處理時,都會向當前Socket對象所在的IO線程信箱發送命令。
讓我們看一眼IOThread對象和IOThreadMailbox的定義
internal sealed class IOThread : ZObject, IMailboxEvent
{
}
IOThread對象繼承自ZObject對象,記得上一節想到ZObject對象知道如何處理各種命令嗎？因此IOThread對象也繼承了他父親的技能。同時IOThread對象實現了IMailboxEvent接口,這個接口之定義了一個方法。
internal interface IMailboxEvent
{
void Ready();
}
當IO信箱接受到命令時表示當前有命令准備好了,可以進行 處理,IO信箱則會調用IO線程的Ready方法處理命令,那么IO信息如何調用IO線程的Ready方法呢，來看下IOThreadMailbox的構造函數。
internal class IOThreadMailbox : IMailbox
{
...
public IOThreadMailbox([NotNull] string name, [NotNull] Proactor proactor, [NotNull] IMailboxEvent mailboxEvent)
{
m_proactor = proactor;
m_mailboxEvent = mailboxEvent;
Command cmd;
bool ok = m_commandPipe.TryRead(out cmd);
}
...
}
在IOThreadMailbox初始化時,傳入了IMailboxEvent。
m_commandPipe是NetMQ的管道(Pipe),后面我們會對其做介紹,這里只要知道該管道用於存放命令即可,可以__暫時__理解為管道隊列。
Proactor
每個IOThread會有一個Proactor,Proactor的工作就是將Socket對象綁定到完成端口,然后定時去掃描完成端口是否有需要處理的Socket對象。
internal class Proactor : PollerBase
{
...
public Proactor([NotNull] string name)
{
m_name = name;
m_stopping = false;
m_stopped = false;
m_completionPort = CompletionPort.Create();
m_sockets = new Dictionary<AsyncSocket, Item>();
}
...
}
Proactor對象繼承自PollerBase,那么PollerBase又是什么呢?從命名可以看這是一個輪詢基類,即該對象需要長時間不斷循環處理某件事情。
PollerBase對象是一個抽象類,它有2個功能：
-
負載均衡
還記的Context中選擇IO線程時有這個一段代碼嗎?
IO線程的負載均衡功能就是PollBase對象提供的每次選擇IO線程時會將
m_load字段值+1protected void AdjustLoad(int amount) { Interlocked.Add(ref m_load, amount); }public int Load { get { #if NETSTANDARD1_3 return Volatile.Read(ref m_load); #else Thread.MemoryBarrier(); return m_load; #endif } }在
IOThread取PollBase對象(Proactor)的Load屬性時候會特殊處理，保證拿到的是最新的值。 -
定時任務
PollBase第二個功能就是支持定時任務，即定時觸發某事件。private readonly SortedList<long, List<TimerInfo>> m_timers;PollBase內部有一個
SortedList,key為任務執行的時間,value為TimeInfo。
TimeInfo對象包含2個信息，id和ITimerEvent接口,id用來辨別當前任務的類型,ITimerEvent接口就包含了TimerEvent方法,即如何執行。
如TcpConnection連接失敗會重新連接時會重連,下面時TcpConnection開始連接方法private void StartConnecting() { Debug.Assert(m_s == null); // Create the socket. try { m_s = AsyncSocket.Create(m_addr.Resolved.Address.AddressFamily, SocketType.Stream, ProtocolType.Tcp); } catch (SocketException) { AddReconnectTimer(); return; } ... } private void AddReconnectTimer() { //獲取重連時間間隔 int rcIvl = GetNewReconnectIvl(); //IO線程的Proactor中,TcpConnection的ReconnectTimerId = 1 m_ioObject.AddTimer(rcIvl, ReconnectTimerId); ... }IO線程會被封裝到
IOObject中,調用IOObject的AddTimer方法實際就是調用IO線程中Proactor對象的AddTimer方法,其方法定義如下public void AddTimer(long timeout, [NotNull] IProactorEvents sink, int id) { long expiration = Clock.NowMs() + timeout; var info = new TimerInfo(sink, id); if (!m_timers.ContainsKey(expiration)) m_timers.Add(expiration, new List<TimerInfo>()); m_timers[expiration].Add(info); }第一行會獲取當前的毫秒時間加上時間間隔。然后加入到
m_timers中。
m_completionPort = CompletionPort.Create();
m_sockets = new Dictionary<AsyncSocket, Item>();
初始化時會創建完成端口，當有socket需要處理時會和完成端口綁定。
初始化時還會初始化一個存放異步AsyncSocket和item的字典。
有關於AsyncSocket和CompletionPort可以去Git上看AsyncIO的源碼,這里不做分析。
Item結構如下
private class Item
{
public Item([NotNull] IProactorEvents proactorEvents)
{
ProactorEvents = proactorEvents;
Cancelled = false;
}
[NotNull]
public IProactorEvents ProactorEvents { get; }
public bool Cancelled { get; set; }
}
它包含了IProactorEvents接口的信息和當前Socket操作是否被取消標志。
internal interface IProactorEvents : ITimerEvent
{
void InCompleted(SocketError socketError, int bytesTransferred);
void OutCompleted(SocketError socketError, int bytesTransferred);
}
IProactorEvents繼承自ITimerEvent。同時它還聲明了InCompleted和OutCompleted方法，即發送或接收完成時如何處理，因此當需要處理Socket時，會將當前Socket處理方式保存到這個字典中。當當前對象發送消息完成,則會調用OutCompleted方法,接收完成時則會調用InCompleted方法。
當有Socket需要綁定時會調用Proactor的AddSocket方法
public void AddSocket(AsyncSocket socket, IProactorEvents proactorEvents)
{
var item = new Item(proactorEvents);
m_sockets.Add(socket, item);
m_completionPort.AssociateSocket(socket, item);
AdjustLoad(1);
}
它包含2個參數,一個時異步Socket對象和IProactorEvents。然后加把他們加入到字段中並將他們綁定到完成端口上。第四段AdjustLoad方法即把當前IO線程處理數量+1，用於負載均衡用。
當Socket操作完成時會調用Proactor的RemoveSocket移除綁定
public void RemoveSocket(AsyncSocket socket)
{
AdjustLoad(-1);
var item = m_sockets[socket];
m_sockets.Remove(socket);
item.Cancelled = true;
}
移除時會將item的Cancelled字段設置為true。所以當Proactor輪詢處理Socket時發現該Socket操作被取消(移除),就會跳過處理。
啟動Procator線程輪詢
在IO線程啟動時實際就是啟動Procator的work線程
public void Start()
{
m_proactor.Start();
}
public void Start()
{
m_worker = new Thread(Loop) { IsBackground = true, Name = m_name };
m_worker.Start();
}
處理socket
完整的Loop方法如下
private void Loop()
{
var completionStatuses = new CompletionStatus[CompletionStatusArraySize];
while (!m_stopping)
{
// Execute any due timers.
int timeout = ExecuteTimers();
int removed;
if (!m_completionPort.GetMultipleQueuedCompletionStatus(timeout != 0 ? timeout : -1, completionStatuses, out removed))
continue;
for (int i = 0; i < removed; i++)
{
try
{
if (completionStatuses[i].OperationType == OperationType.Signal)
{
var mailbox = (IOThreadMailbox)completionStatuses[i].State;
mailbox.RaiseEvent();
}
// if the state is null we just ignore the completion status
else if (completionStatuses[i].State != null)
{
var item = (Item)completionStatuses[i].State;
if (!item.Cancelled)
{
switch (completionStatuses[i].OperationType)
{
case OperationType.Accept:
case OperationType.Receive:
item.ProactorEvents.InCompleted(
completionStatuses[i].SocketError,
completionStatuses[i].BytesTransferred);
break;
case OperationType.Connect:
case OperationType.Disconnect:
case OperationType.Send:
item.ProactorEvents.OutCompleted(
completionStatuses[i].SocketError,
completionStatuses[i].BytesTransferred);
break;
default:
throw new ArgumentOutOfRangeException();
}
}
}
}
catch (TerminatingException)
{ }
}
}
}
var completionStatuses = new CompletionStatus[CompletionStatusArraySize];
第一行初始化了CompletionStatus數組,CompletionStatusArraySize值為100。
CompletionStatus作用是用來保存socket的信息或狀態。
獲取超時時間
int timeout = ExecuteTimers();
protected int ExecuteTimers()
{
if (m_timers.Count == 0)
return 0;
long current = Clock.NowMs();
var keys = m_timers.Keys;
for (int i = 0; i < keys.Count; i++)
{
var key = keys[i];
if (key > current)
{
return (int)(key - current);
}
var timers = m_timers[key];
foreach (var timer in timers)
{
timer.Sink.TimerEvent(timer.Id);
}
timers.Clear();
m_timers.Remove(key);
i--;
}
return 0;
}
ExecuteTimers會計算之前加入到m_timers需要等待的超時時間,若沒有對象則直接返回0,否則獲取若獲取到key時間在當前時間之前,則需要調用TimerEvent方法，調用完成后移除。
若獲取到的key時間比當前時間大,則返回他們的差即為需要等待的超時時間。
從完成端口獲取處理完的狀態
int removed;
if (!m_completionPort.GetMultipleQueuedCompletionStatus(timeout != 0 ? timeout : -1, completionStatuses, out removed))
continue;
GetMultipleQueuedCompletionStatus方法傳入一個超時時間,若前面獲取的超時時間為0,則這邊會設置為-1,表示阻斷直到有要處理的才返回。
CompletionPort內部維護了一個狀態隊列，removed即為處理完成返回的狀態個數。
若獲取成功則會返回true,后面就開始遍歷completionStatuses數組處理完成Socket。
開始處理待處理的狀態
public struct CompletionStatus
{
internal CompletionStatus(AsyncSocket asyncSocket, object state, OperationType operationType, SocketError socketError, int bytesTransferred) :
this()
{
AsyncSocket = asyncSocket;
State = state;
OperationType = operationType;
SocketError = socketError;
BytesTransferred = bytesTransferred;
}
public AsyncSocket AsyncSocket { get; private set; }
public object State { get; internal set; }
public OperationType OperationType { get; internal set; }
public SocketError SocketError { get; internal set; }
public int BytesTransferred { get; internal set; }
}
CompletionStatus是個結構體,它包含的信息如上。其中OperationType是當前Socket的處理方式。
public enum OperationType
{
Send, Receive, Accept, Connect, Disconnect, Signal
}
在for循環的一開始先會判斷當前狀態的OperationType,若是Signal,則說明當前是個信號狀態,說明有命令需要處理,則會調用IO信箱的RaiseEvent方法,實際為IO線程的Ready方法。
public void Ready()
{
Command command;
while (m_mailbox.TryRecv(out command))
command.Destination.ProcessCommand(command);
}
IOThread會將當前信箱的所有命令進行處理。
若不是Signal則會將CompletionStatus保存的狀態信息轉換為Item對象,並判斷當前Socket是否移除(取消)。若沒有則對其進行處理。判斷OperationType,若為Accept或Receive則表示需要接收,則調用InCompleted方法。若為Connect,Disconnect或Send則表示有消息向外發送,則調用OutCompleted方法。
至此IOThread代碼分析完畢。
IOObject
internal class IOObject : IProactorEvents
{
public IOObject([CanBeNull] IOThread ioThread)
{
if (ioThread != null)
Plug(ioThread);
}
public void Plug([NotNull] IOThread ioThread)
{
Debug.Assert(ioThread != null);
m_ioThread = ioThread;
}
}
IOObject實際就是保存了IOThread的信息和Socket處理完成時如何執行,以及向外暴露了一些接口。
再次說明,如果向簡單了解完成端口如何使用,則看《完成端口使用》，如果想詳細了解完成端口則看下《完成端口詳細介紹》,如果想直到NetMQ的AsyncIO和完成端口的源碼請看AsyncIO。
總結
該篇介紹了IO線程和完成端口的處理方式,若哪里分析的不到位或有誤希望支出。
微信掃一掃二維碼關注訂閱號傑哥技術分享
本文地址：https://www.cnblogs.com/Jack-Blog/p/6347163.html
作者博客：傑哥很忙
歡迎轉載，請在明顯位置給出出處及鏈接)
