消息隊列NetMQ 原理分析5-StreamEngine、Encord和Decord

• 消息隊列NetMQ 原理分析5-StreamEngine,Encord和Decord
  • 前言
    • 介紹
    • 目的
  • StreamEngine
    • 發送數據
    • 接收數據
    • 流程分析
  • Encoder
    • V2Encoder
    • V1Encoder
    • RawEncoder
  • Decoder
    • V2Decoder
    • V1Decoder
    • RawDecoder
  • 總結

前言介紹

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

目的

對NetMQ的源碼進行學習並分析理解,因此寫下該系列文章,本系列文章暫定編寫計劃如下：

1. 消息隊列NetMQ 原理分析1-Context和ZObject
2. 消息隊列NetMQ 原理分析2-IO線程和完成埠
3. 消息隊列NetMQ 原理分析3-命令產生/處理、創建Socket和回收線程
4. 消息隊列NetMQ 原理分析4-Socket、Session、Option和Pipe
5. 消息隊列NetMQ 原理分析5-StreamEngine,Encord和Decord
6. 消息隊列NetMQ 原理分析6-TCP和Inpoc實現
7. 消息隊列NetMQ 原理分析7-Device
8. 消息隊列NetMQ 原理分析8-不同類型的Socket
9. 消息隊列NetMQ 原理分析9-實戰

友情提示: 看本系列文章時最好獲取源碼,更有助於理解。

StreamEngine

SocketBase將Msg發送給SessionBase之後需要將Msg轉化為byte進行傳輸,Engine就是做轉換的工作,轉換完成之後就會和實際的底層Socket進行消息傳輸。

NetMQ在Tcp協議消息轉換使用的是StreamEngine。

internal sealed class StreamEngine : IEngine, IProactorEvents, IMsgSink
{

}

上一章介紹到管道事件。

當出管道有數據可讀時，會調用SessionBase的ReadActivated事件

public void ReadActivated(Pipe pipe)
{
...
if (m_engine != null)
m_engine.ActivateOut;
else
m_pipe.CheckRead;
}

然後會調用對應m_engine的ActivateOut事件

public void ActivateOut
{
FeedAction(Action.ActivateOut, SocketError.Success, 0);
}

BeginSending;
}
break;
...
}
...
}

當TCPConnect客戶端發送請求完成時，會調用OutCompleted事件

private void Loop
{
...
switch (completion.OperationType)
{
...
case OperationType.Connect:
case OperationType.Disconnect:
case OperationType.Send:
item.ProactorEvents.OutCompleted(
completion.SocketError,
completion.BytesTransferred);
}
}
...

public void OutCompleted(SocketError socketError, int bytesTransferred)
{
...
// Create the engine object for this connection.
var engine = new StreamEngine(m_s, m_options, m_endpoint);
...
// Attach the engine to the corresponding session object.
SendAttach(m_session, engine);
...
}

此時會創建一個StreamEngine和請求的SessionBase對象進行關聯。

protected override void ProcessAttach(IEngine engine)
{
Debug.Assert(engine != null);

// Plug in the engine.
Debug.Assert(m_engine == null);
m_engine = engine;
m_engine.Plug(m_ioThread, this);
}

接收數據

當完成埠通知數據接收完成時，會調用Proactor的InCompleted事件，實際就是調用的對應的StreamEngine的InCompleted事件

public void InCompleted(SocketError socketError, int bytesTransferred)
{
FeedAction(Action.InCompleted, socketError, bytesTransferred);
}

public void FeedAction{
...
case State.Active:
switch (action)
{
case Action.InCompleted:
m_insize = EndRead(socketError, bytesTransferred);

ProcessInput;
break;
...
}
...
}

接收完成後會對接收到的數據進行處理

private void ProcessInput
{
...
if (m_options.RawSocket)
{
if (m_insize == 0 || !m_decoder.MessageReadySize(m_insize))
{
processed = 0;
}
else
{
processed = m_decoder.ProcessBuffer(m_inpos, m_insize);
}
}
else
{
// Push the data to the decoder.
processed = m_decoder.ProcessBuffer(m_inpos, m_insize);
}
...
// Flush all messages the decoder may have produced.
m_session.Flush;
...
}
public override bool MessageReadySize(int msgSize)
{
m_inProgress = new Msg;
m_inProgress.InitPool(msgSize);

return true;
}

讀取數據到Msg後會調用Decoder的ProcessBuffer方法

PS:由於NetMQ有自己的傳輸協議格式，因此當使用NetMQ和其他程序進行Socket傳輸時，必須使用StreamSocket。


public int ProcessBuffer(ByteArraySegment data, int size)
{
...
while (m_toRead == 0)
{
if (!Next)
{
if (State < 0)
{
return -1;
}
return size;
}
}
return size;
...
}
protected override bool Next
{
if (State == RawMessageReadyState)
{
return RawMessageReady;
}

if (isMessagedPushed)
{
// NOTE: This is just to break out of process_buffer
// raw_message_ready should never get called in state machine w/o
// message_ready_size from stream_engine.
NextStep(new ByteArraySegment(m_inProgress.Data, m_inProgress.Offset),
1, RawMessageReadyState);
}
return isMessagedPushed;
...
}

對讀到的數據進行處理調用RawDecoder的Next的方法,將獲取到的Msg放入到SeesionBase的管道中。

讀寫數據流程圖如下圖所示：

我們使用WireShark進行驗證。

我們監聽15557地址，然後創建一個客戶端連接15557地址

前面3條是三次握手。第四條是客戶端向伺服器發送了10位元組長度的請求頭部，以0xff開頭，0x7f結尾。中間是8位元組是Identitysize長度

...
switch (m_handshakeState)
{
case HandshakeState.Closed:
switch (action)
{
case Action.Start:
// Send the "length" and "flags" fields of the identity message.
// The "length" field is encoded in the long format.
m_greetingOutputBuffer[m_outsize++] = 0xff;
m_greetingOutputBuffer.PutLong(m_options.Endian, (long)m_options.IdentitySize + 1, 1);
m_outsize += 8;
m_greetingOutputBuffer[m_outsize++] = 0x7f;
...
}
...
}
...

第6條是伺服器向客戶端發送的10位元組長度的請求頭部，以0xff開頭，0x7f結尾。中間是8位元組是identitysize的信息

第8條是伺服器向客戶端發送的版本號和Socket類型,01表示版本號1，06表示當前是RouterSocket

...
case HandshakeState.ReceivingGreeting:
switch (action)
{
case Action.InCompleted:
...

if (m_greeting[0] != 0xff || (m_greetingBytesRead == 10 && (m_greeting[9] & 0x01) == 0)){
...
}
else if (m_greetingBytesRead < 10)
{
var greetingSegment = new ByteArraySegment(m_greeting, m_greetingBytesRead);
BeginRead(greetingSegment, PreambleSize - m_greetingBytesRead);
}
else
{
...
m_outpos[m_outsize++] = 1; // Protocol version
m_outpos[m_outsize++] = (byte)m_options.SocketType;
...
}
...
}
...

第10條是客戶端向伺服器發送的版本號和socket類型,05表示當前是DealSocket

...
case HandshakeState.ReceivingRestOfGreeting:
switch (action)
{
case Action.InCompleted:
...
if (m_greeting[VersionPos] == 0)
{
// ZMTP/1.0 framing.
m_encoder = new V1Encoder(Config.OutBatchSize, m_options.Endian);
m_encoder.SetMsgSource(m_session);

m_decoder = new V1Decoder(Config.InBatchSize, m_options.MaxMessageSize, m_options.Endian);
m_decoder.SetMsgSink(m_session);
}
else
{
// v1 framing protocol.
m_encoder = new V2Encoder(Config.OutBatchSize, m_session, m_options.Endian);
m_decoder = new V2Decoder(Config.InBatchSize, m_options.MaxMessageSize, m_session, m_options.Endian);
}
Activate;
...
}
...

EncoderV2Encoder

接下來就是數據傳輸。

public V2Encoder(int bufferSize, IMsgSource session, Endianness endian)
: base(bufferSize, endian)
{
m_inProgress = new Msg;
m_inProgress.InitEmpty;

// Write 0 bytes to the batch and go to message_ready state.
NextStep(m_tmpbuf, 0, MessageReadyState, true);
}

由於NetMQ使用的是版本1，用的是V2Encoder和V2Decoder進行編碼和解碼。

在初始化Encoder的時候會向報文寫入2個0位元組數據，暫時不明白為何要這樣做。


int protocolFlags = 0;
if (m_inProgress.HasMore)
protocolFlags |= V2Protocol.MoreFlag;
if (m_inProgress.Size > 255)
protocolFlags |= V2Protocol.LargeFlag;
m_tmpbuf[0] = (byte)protocolFlags;

// Encode the message length. For messages less then 256 bytes,
// the length is encoded as 8-bit unsigned integer. For larger
// messages, 64-bit unsigned integer in network byte order is used.
int size = m_inProgress.Size;
if (size > 255)
{
m_tmpbuf.PutLong(Endian, size, 1);
NextStep(m_tmpbuf, 9, SizeReadyState, false);
}
else
{
m_tmpbuf[1] = (byte)(size);
NextStep(m_tmpbuf, 2, SizeReadyState, false);
}

第一個位元組是Flags用於標記該報文是否為大報文，超過過255個位元組就會標記為大包標記，是否還有更多報文。若報文長度小於256，則第二個位元組用於存儲報文長度。但是若是大報文，則會8個位元組保存報文長度。

下面就開始發送數據

我們用客戶端發一個字元串test1，然後服務端原樣返回該字元串

可以看到如我們上面分析的一樣，第一個位元組為0，第二個位元組為大小test1為5個位元組長度。由於CMD命令單行輸入最長字元限制長度為255，因此我們沒辦法在CMD命令下輸入更長數據進行測試。暫時就不做驗證。

V1Encoder編碼如下所示

if (size < 255)
{
m_tmpbuf[0] = (byte)size;
m_tmpbuf[1] = (byte)(m_inProgress.Flags & MsgFlags.More);
NextStep(m_tmpbuf, 2, SizeReadyState, false);
}
else
{
m_tmpbuf[0] = 0xff;
m_tmpbuf.PutLong(Endian, size, 1);
m_tmpbuf[9] = (byte)(m_inProgress.Flags & MsgFlags.More);
NextStep(m_tmpbuf, 10, SizeReadyState, false);
}

當小於255字元，首字元是長度，第二個字元是Flags，超過255字元，首字元為0xff,然後跟著8個字元長度的長度值，接下來是Flags

使用RawEncoder會將原始數據原樣發送不會增加任何其他字元。

接收到數據會先接收第一個位元組Flags判斷是否有後續包以及是小包還是打包，若是小包，則解析第一個位元組長度位，否則讀取8個位元組長度位。

接收到數據收先會判斷第一個位元組是不是Oxff，若為Oxff則表示為打包，獲取8位位元組長度，否則獲取1位位元組長度處理。

使用RawDecoder會讀取數據保存到管道中。

本片介紹了NetMQ的報文格式並闡述了底層Msg如何轉換為流進行發送和接收。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！