[关闭]
@a604572782 2017-08-04T11:15:00.000000Z 字数 8082 阅读 2300

消息队列NetMQ 原理分析5-StreamEngine,Encord和Decord

NetMQ ZeroMQ C#

前言

介绍

[NetMQ](https://github.com/zeromq/netmq.git)是ZeroMQ的C#移植版本,它是对标准socket接口的扩展。它提供了一种异步消息队列,多消息模式,消息过滤（订阅）,对多种传输协议的无缝访问。
当前有2个版本正在维护，版本3最新版为3.3.4，版本4最新版本为4.0.1。本文档是对4.0.1分支代码进行分析。

zeromq的英文文档
NetMQ的英文文档

目的

对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进行消息传输。
201753102922-2
NetMQ在Tcp协议消息转换使用的是StreamEngine。

  1. internal sealed class StreamEngine : IEngine, IProactorEvents, IMsgSink
  2. {
  4. }

上一章介绍到管道事件。

发送数据

当出管道有数据可读时，会调用SessionBase的ReadActivated事件

  1. public void ReadActivated(Pipe pipe)
  2. {
  3.     ...
  4.     if (m_engine != null)
  5.         m_engine.ActivateOut();
  6.     else
  7.         m_pipe.CheckRead();
  8. }

然后会调用对应m_engine的ActivateOut事件

  1. public void ActivateOut()
  2. {
  3.     FeedAction(Action.ActivateOut, SocketError.Success, 0);
  4. }
  6. public void FeedAction(){
  7.     ...
  8.     case State.Active:
  9.         switch (action)
  10.         {
  11.             case Action.OutCompleted:
  12.                 int bytesSent = EndWrite(socketError, bytesTransferred);
  14.                 // IO error has occurred. We stop waiting for output events.
  15.                 // The engine is not terminated until we detect input error;
  16.                 // this is necessary to prevent losing incoming messages.
  17.                 if (bytesSent == -1)
  18.                 {
  19.                     m_sendingState = SendState.Error;
  20.                 }
  21.                 else
  22.                 {
  23.                     m_outpos.AdvanceOffset(bytesSent);
  24.                     m_outsize -= bytesSent;
  26.                     BeginSending();
  27.                 }
  28.                 break;
  29.             ...
  30.         }
  31.     ...
  32. }

当TCPConnect客户端发送请求完成时，会调用OutCompleted事件

  1. private void Loop()
  2. {
  3.     ...
  4.     switch (completion.OperationType)
  5.     {
  6.         ...
  7.         case OperationType.Connect:
  8.         case OperationType.Disconnect:
  9.         case OperationType.Send:
  10.         item.ProactorEvents.OutCompleted(
  11.             completion.SocketError,
  12.             completion.BytesTransferred);
  13.             }
  14.     }
  15.     ...
  1. public void OutCompleted(SocketError socketError, int bytesTransferred)
  2. {
  3.     ...
  4.     // Create the engine object for this connection.
  5.     var engine = new StreamEngine(m_s, m_options, m_endpoint);
  6.     ...
  7.     // Attach the engine to the corresponding session object.
  8.     SendAttach(m_session, engine);
  9.     ...
  10.     }

此时会创建一个StreamEngine和请求的SessionBase对象进行关联。

  1. protected override void ProcessAttach(IEngine engine)
  2. {
  3.     Debug.Assert(engine != null);
  5.     // Create the pipe if it does not exist yet.
  6.     if (m_pipe == null && !IsTerminating)
  7.     {
  8.         ZObject[] parents = { this, m_socket };
  9.         int[] highWaterMarks = { m_options.ReceiveHighWatermark, m_options.SendHighWatermark };
  10.         int[] lowWaterMarks = { m_options.ReceiveLowWatermark, m_options.SendLowWatermark };
  11.         bool[] delays = { m_options.DelayOnClose, m_options.DelayOnDisconnect };
  12.         Pipe[] pipes = Pipe.PipePair(parents, highWaterMarks, lowWaterMarks, delays);
  14.         // Plug the local end of the pipe.
  15.         pipes[0].SetEventSink(this);
  17.         // Remember the local end of the pipe.
  18.         Debug.Assert(m_pipe == null);
  19.         m_pipe = pipes[0];
  21.         // Ask socket to plug into the remote end of the pipe.
  22.         SendBind(m_socket, pipes[1]);
  23.     }
  25.     // Plug in the engine.
  26.     Debug.Assert(m_engine == null);
  27.     m_engine = engine;
  28.     m_engine.Plug(m_ioThread, this);
  29. }

接收数据

当完成端口通知数据接收完成时，会调用Proactor的InCompleted事件，实际就是调用的对应的StreamEngine的InCompleted事件

  1. public void InCompleted(SocketError socketError, int bytesTransferred)
  2. {
  3.     FeedAction(Action.InCompleted, socketError, bytesTransferred);
  4. }
  1. public void FeedAction(){
  2.     ...
  3.     case State.Active:
  4.         switch (action)
  5.         {
  6.             case Action.InCompleted:
  7.                 m_insize = EndRead(socketError, bytesTransferred);
  9.                 ProcessInput();
  10.                 break;
  11.             ...
  12.         }
  13.     ...
  14. }

接收完成后会对接收到的数据进行处理

  1. private void ProcessInput()
  2. {
  3.     ...
  4.     if (m_options.RawSocket)
  5.     {
  6.         if (m_insize == 0 || !m_decoder.MessageReadySize(m_insize))
  7.         {
  8.             processed = 0;
  9.         }
  10.         else
  11.         {
  12.             processed = m_decoder.ProcessBuffer(m_inpos, m_insize);
  13.         }
  14.     }
  15.     else
  16.     {
  17.         // Push the data to the decoder.
  18.         processed = m_decoder.ProcessBuffer(m_inpos, m_insize);
  19.     }
  20.     ...
  21.     // Flush all messages the decoder may have produced.
  22.     m_session.Flush();
  23.     ...
  24. }
  25. public override bool MessageReadySize(int msgSize)
  26. {
  27.     m_inProgress = new Msg();
  28.     m_inProgress.InitPool(msgSize);
  30.     NextStep(new ByteArraySegment(m_inProgress.Data, m_inProgress.Offset),
  31.         m_inProgress.Size, RawMessageReadyState);
  33.     return true;
  34. }

读取数据到Msg后会调用Decoder的ProcessBuffer方法

PS:由于NetMQ有自己的传输协议格式，因此当使用NetMQ和其他程序进行Socket传输时，必须使用StreamSocket。

  2. public int ProcessBuffer(ByteArraySegment data, int size)
  3. {
  4.     ...
  5.     while (m_toRead == 0)
  6.     {
  7.         if (!Next())
  8.         {
  9.             if (State < 0)
  10.             {
  11.                 return -1;
  12.             }
  13.             return size;
  14.         }
  15.     }
  16.     return size;
  17.     ...
  18. }
  19. protected override bool Next()
  20. {
  21.     if (State == RawMessageReadyState)
  22.     {
  23.         return RawMessageReady();
  24.     }
  26.     return false;
  27. }
  28. private bool RawMessageReady()
  29. {
  30.     ...
  31.     bool isMessagedPushed = m_msgSink.PushMsg(ref m_inProgress);
  33.     if (isMessagedPushed)
  34.     {
  35.         // NOTE: This is just to break out of process_buffer
  36.         // raw_message_ready should never get called in state machine w/o
  37.         // message_ready_size from stream_engine.
  38.         NextStep(new ByteArraySegment(m_inProgress.Data, m_inProgress.Offset),
  39.             1, RawMessageReadyState);
  40.     }
  41.     return isMessagedPushed;
  42.     ...
  43. }

对读到的数据进行处理调用RawDecoder的Next的方法,将获取到的Msg放入到SeesionBase的管道中。

流程分析

读写数据流程图如下图所示：
2017731191255-StreamEngine

我们使用WireShark进行验证。

我们监听15557地址，然后创建一个客户端连接15557地址
2017731192245-1
前面3条是三次握手。第四条是客户端向服务器发送了10字节长度的请求头部，以0xff开头，0x7f结尾。中间是8字节是Identitysize长度

  1. ...
  2. switch (m_handshakeState)
  3. {
  4.     case HandshakeState.Closed:
  5.         switch (action)
  6.         {
  7.             case Action.Start:
  8.                 // Send the 'length' and 'flags' fields of the identity message.
  9.                 // The 'length' field is encoded in the long format.
  10.                 m_greetingOutputBuffer[m_outsize++] = 0xff;
  11.                 m_greetingOutputBuffer.PutLong(m_options.Endian, (long)m_options.IdentitySize + 1, 1);
  12.                 m_outsize += 8;
  13.                 m_greetingOutputBuffer[m_outsize++] = 0x7f;
  14.                 ...
  15.         }
  16.         ...
  17. }
  18. ...

201773119305-2
第6条是服务器向客户端发送的10字节长度的请求头部，以0xff开头，0x7f结尾。中间是8字节是identitysize的信息
2017731194159-3
第8条是服务器向客户端发送的版本号和Socket类型,01表示版本号1，06表示当前是RouterSocket

  1. ...
  2. case HandshakeState.ReceivingGreeting:
  3.     switch (action)
  4.     {
  5.         case Action.InCompleted:
  6.         ...
  8.                 if (m_greeting[0] != 0xff || (m_greetingBytesRead == 10 && (m_greeting[9] & 0x01) == 0)){
  9.                 ...
  10.                 }
  11.                 else if (m_greetingBytesRead < 10)
  12.                 {
  13.                     var greetingSegment = new ByteArraySegment(m_greeting, m_greetingBytesRead);
  14.                     BeginRead(greetingSegment, PreambleSize - m_greetingBytesRead);
  15.                 }
  16.                 else
  17.                 {
  18.                     ...
  19.                     m_outpos[m_outsize++] = 1; // Protocol version
  20.                     m_outpos[m_outsize++] = (byte)m_options.SocketType;
  21.                     ...
  22.                 }
  23.         ...
  24.     }
  25. ...

2017731194312-4
第10条是客户端向服务器发送的版本号和socket类型,05表示当前是DealSocket

  1. ...
  2. case HandshakeState.ReceivingRestOfGreeting:
  3.     switch (action)
  4.     {
  5.         case Action.InCompleted:
  6.         ...
  7.         if (m_greeting[VersionPos] == 0)
  8.         {
  9.             // ZMTP/1.0 framing.
  10.             m_encoder = new V1Encoder(Config.OutBatchSize, m_options.Endian);
  11.             m_encoder.SetMsgSource(m_session);
  13.             m_decoder = new V1Decoder(Config.InBatchSize, m_options.MaxMessageSize, m_options.Endian);
  14.             m_decoder.SetMsgSink(m_session);
  15.         }
  16.         else
  17.         {
  18.             // v1 framing protocol.
  19.             m_encoder = new V2Encoder(Config.OutBatchSize, m_session, m_options.Endian);
  20.             m_decoder = new V2Decoder(Config.InBatchSize, m_options.MaxMessageSize, m_session, m_options.Endian);
  21.         }
  22.         Activate();
  23.         ...
  24.     }
  25. ...

Encoder

V2Encoder

接下来就是数据传输。

  1. public V2Encoder(int bufferSize, IMsgSource session, Endianness endian)
  2.     : base(bufferSize, endian)
  3. {
  4.     m_inProgress = new Msg();
  5.     m_inProgress.InitEmpty();
  7.     m_msgSource = session;
  9.     // Write 0 bytes to the batch and go to message_ready state.
  10.     NextStep(m_tmpbuf, 0, MessageReadyState, true);
  11. }

由于NetMQ使用的是版本1，用的是V2Encoder和V2Decoder进行编码和解码。
在初始化Encoder的时候会向报文写入2个0字节数据，暂时不明白为何要这样做。
2017731194443-5

  2. int protocolFlags = 0;
  3. if (m_inProgress.HasMore)
  4.     protocolFlags |= V2Protocol.MoreFlag;
  5. if (m_inProgress.Size > 255)
  6.     protocolFlags |= V2Protocol.LargeFlag;
  7. m_tmpbuf[0] = (byte)protocolFlags;
  9. // Encode the message length. For messages less then 256 bytes,
  10. // the length is encoded as 8-bit unsigned integer. For larger
  11. // messages, 64-bit unsigned integer in network byte order is used.
  12. int size = m_inProgress.Size;
  13. if (size > 255)
  14. {
  15.     m_tmpbuf.PutLong(Endian, size, 1);
  16.     NextStep(m_tmpbuf, 9, SizeReadyState, false);
  17. }
  18. else
  19. {
  20.     m_tmpbuf[1] = (byte)(size);
  21.     NextStep(m_tmpbuf, 2, SizeReadyState, false);
  22. }

第一个字节是Flags用于标记该报文是否为大报文，超过过255个字节就会标记为大包标记，是否还有更多报文。若报文长度小于256，则第二个字节用于存储报文长度。但是若是大报文，则会8个字节保存报文长度。
下面就开始发送数据
我们用客户端发一个字符串test1，然后服务端原样返回该字符串
201784102144-6
可以看到如我们上面分析的一样，第一个字节为0，第二个字节为大小test1为5个字节长度。由于CMD命令单行输入最长字符限制长度为255，因此我们没办法在CMD命令下输入更长数据进行测试。暂时就不做验证。

V1Encoder

V1Encoder编码如下所示

  1. if (size < 255)
  2. {
  3.     m_tmpbuf[0] = (byte)size;
  4.     m_tmpbuf[1] = (byte)(m_inProgress.Flags & MsgFlags.More);
  5.     NextStep(m_tmpbuf, 2, SizeReadyState, false);
  6. }
  7. else
  8. {
  9.     m_tmpbuf[0] = 0xff;
  10.     m_tmpbuf.PutLong(Endian, size, 1);
  11.     m_tmpbuf[9] = (byte)(m_inProgress.Flags & MsgFlags.More);
  12.     NextStep(m_tmpbuf, 10, SizeReadyState, false);
  13. }

当小于255字符，首字符是长度，第二个字符是Flags，超过255字符，首字符为0xff,然后跟着8个字符长度的长度值，接下来是Flags

RawEncoder

使用RawEncoder会将原始数据原样发送不会增加任何其他字符。

Decoder

V2Decoder

接收到数据会先接收第一个字节Flags判断是否有后续包以及是小包还是打包，若是小包，则解析第一个字节长度位，否则读取8个字节长度位。

V1Decoder

接收到数据收先会判断第一个字节是不是Oxff，若为Oxff则表示为打包，获取8位字节长度，否则获取1位字节长度处理。

RawDecoder

使用RawDecoder会读取数据保存到管道中。

总结

本片介绍了NetMQ的报文格式并阐述了底层Msg如何转换为流进行发送和接收。

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注