消息隊列NetMQ 原理分析2-IO線程和完成端口

前言
- 介紹
- 目的
IO線程
初始化IO線程
- Proactor
- 啟動Procator線程輪詢
- 處理socket
IOObject
總結

前言

介紹

[NetMQ](https://github.com/zeromq/netmq.git)是ZeroMQ的C#移植版本,它是對標準socket接口的擴展。它提供了一種異步消息隊列,多消息模式,消息過濾（訂閱）,對多種傳輸協議的無縫訪問。
當前有2個版本正在維護，版本3最新版為3.3.4，版本4最新版本為4.0.0-rc5。本文檔是對4.0.0-rc5分支代碼進行分析。

zeromq的英文文檔
NetMQ的英文文檔

目的

對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字段值+1
protected 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 statuselse 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處理完成時如何執行,以及向外暴露了一些接口。