[关闭]
@a604572782 2017-04-27T16:35:01.000000Z 字数 5336 阅读 2484

消息队列NetMQ 原理分析1-Context和ZObject

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-Session、Option和Pipe
5. 消息队列NetMQ 原理分析5-Engine
6. 消息队列NetMQ 原理分析6-TCP和Inpoc实现
7. 消息队列NetMQ 原理分析7-Device
8. 消息队列NetMQ 原理分析8-不同类型的Socket
9. 消息队列NetMQ 原理分析9-实战

友情提示: 看本系列文章时最好获取源码,更有助于理解。

Context

NetMQ有一个Context对象,用于初始化并保存当前NetMQ底层的对象状态,如IO线程、回收线程、进程间传输节点字典、插槽m_slots(用于保存IO对象，回收对象和socket对象的Mailbox)、初始化但未用到的Socket对象指针数组以及当前Mailbox(用于接收终止信号)等。

初始化Context

当创建第一个Socket对象时会初始化IO线程，回收线程以及工作线程。默认Socket数量1024个,IO线程1个,回收线程1个。

  1. m_slots = new Mailbox[m_slotCount];//m_soltCount = 1 + 1 + 1024

m_slots[0]保存的是Context的Mailbox。

  1. m_slots[TermTid] = m_termMailbox;//用于当前Context接收终止信号

m_slots[1]保存的是回收对象的Mailbox,保存完毕后就会启动回收对象轮询线程。

  1. m_reaper = new Reaper(this, ReaperTid);//ReaperTid = 1
  2. m_slots[ReaperTid] = m_reaper.Mailbox;
  3. m_reaper.Start();

m_slots[2]保存的是IO线程对象的Mailbox。

  1. for (int i = 2; i != ios/*ios = 1,默认用1个io线程*/ + 2; i++)
  2. {
  3.     IOThread ioThread = new IOThread(this, i);
  4.     m_ioThreads.Add(ioThread);
  5.     m_slots[i] = ioThread.Mailbox;
  6.     ioThread.Start();
  7. }

其余1024个slot保存的是socket对象的Mailbox,当socket还没使用是,slots保存的是null,占个位置,同时m_emptySlots。

  1. //m_soltCount = 1 + 1 + 1024
  2. for (int i = (int)m_slotCount - 1; >= (int)ios + 2; i--)
  3. {
  4.     m_emptySlots.Push(i);
  5.     m_slots[i] = null;
  6. }

创建SocketBase

无论是什么类型的Socket全都是在Context中进行创建或释放的。NetMQ中不同Socket都继承自SocketBase,在Context未中止且Socket未满时,会从m_emptySlots栈中Pop出一个未使用的指针。若创建失败,则重新加回到栈中,否则更新当前使用的Socket的集合加入该Socket并更新m_slots的Mailbox

  1. //slot是当前socket在s_slots中的位置,也用于生成SocketBase的`ThreadId`
  2. int slot = m_emptySlots.Pop();
  3. //  sid是生成并递增的唯一的socket ID,用于SocketBase创建MailBox命名用,并无实际其他作用。
  4. int sid = Interlocked.Increment(ref s_maxSocketId);
  5. s = SocketBase.Create(type, this, slot, sid);
  6. if (s == null)
  7. {
  8.     m_emptySlots.Push(slot);
  9.     return null;
  10. }
  11. m_sockets.Add(s);
  12. m_slots[slot] = s.Mailbox;

释放SocketBase

当Reaper要释放某个SocketBase时,最终会调用Context的DestroySocket方法。

  1. tid = socket.ThreadId;
  2. //重新加入到可用socket栈中
  3. m_emptySlots.Push(tid);
  4. //关闭连接
  5. m_slots[tid].Close();
  6. //清空引用
  7. m_slots[tid] = null;
  8. //  从当前使用socket集合移除
  9. m_sockets.Remove(socket);
  10. //若当前接收到中止信号且当前socket全部已释放时停止回收线程
  11. if (m_terminating && m_sockets.Count == 0)
  12.     m_reaper.Stop();

缓存进程内通信Socket

NetMQ除了支持TCP以外还支持inproc(进程内通讯),ipc(进程间通讯),pgm和epgm(多路广播)等传输协议。
Context会用一个字典管理当前使用inpoc的socket。
当inpoc的socket进行绑定时会加入到字典缓存中。释放时会从字典缓存中移除。当使用inpoc协议连接时,增加当前绑定inpoc地址的连接数。

ZObject

ZObject是NetMQ的Session(状态),IOThread(IO线程),Repear(回收线程),Pipe(管道),Own(所属关系)对象的基类,它是包含2个信息,当前全局Context对象,以及当前对象处理的线程Id。所有socket最终都是继承自该对象。因此ZObject对象需要知道IO对象接收到不同命令时如何进行处理命令。
NetMQ中一共定义了一下的命令类型

  1. public enum CommandType
  2. {
  3.     //  发送给IO线程表示当前对象需要停止
  4.     Stop,
  5.     //  发送给IO线程表示当前对象需要注册到IO线程中
  6.     Plug,
  7.     //  将创建的对象Session的加入到当前Socket的所属集合中
  8.     Own,
  9.     //  附加engine到Session中
  10.     Attach,
  11.     //  建立session到Socket之间的管道,在握手之前调用inc_seqnum.
  12.     Bind,
  13.     //  通过写管道发送通知给读管道多少信息可读
  14.     ActivateRead,
  15.     //  通过读管道发送通知给写读管道多少信息可写
  16.     ActivateWrite,
  17.     //  创建一个新的管道后通过读管道发送给写管道
  18.     //  参数是管道类型,然而,他的目的地是私有的,因此我们必须用void指针, however,
  19.     Hiccup,
  20.     //  通过读管道发送到写管道告诉他中止所有管道
  21.     PipeTerm,
  22.     //  写管道对PipeTerm命令响应
  23.     PipeTermAck,
  24.     //  通过IO对象发送给socket请求终端IO对象
  25.     TermReq,
  26.     //  通过socket发送给IO对象他自己开始关闭
  27.     Term,
  28.     //  通过IO对象发送给socket让它知道已经关闭
  29.     TermAck,
  30.     //  将关闭套接字的所有权转移给回收线程.
  31.     Reap,
  32.     //  关闭套接字通知回收线程他已经释放
  33.     Reaped,
  34.     //  当所有socket都被释放通过回收线程发送给 term 线程
  35.     Done
  36. }

根据不同命令类型进行处理,处理方式由具体的Socket子类去重载。

  1. public void ProcessCommand(Command cmd)
  2. {
  3.     switch (cmd.CommandType)
  4.     {
  5.         case CommandType.ActivateRead:
  6.             ProcessActivateRead();
  7.             break;
  8.         case CommandType.ActivateWrite:
  9.             ProcessActivateWrite((long)cmd.Arg);
  10.             break;
  11.         case CommandType.Stop:
  12.             ProcessStop();
  13.             break;
  14.         case CommandType.Plug:
  15.             ProcessPlug();
  16.             ProcessSeqnum();
  17.             break;
  18.         case CommandType.Own:
  19.             ProcessOwn((Own)cmd.Arg);
  20.             ProcessSeqnum();
  21.             break;
  22.         case CommandType.Attach:
  23.             ProcessAttach((IEngine)cmd.Arg);
  24.             ProcessSeqnum();
  25.             break;
  26.         case CommandType.Bind:
  27.             ProcessBind((Pipe)cmd.Arg);
  28.             ProcessSeqnum();
  29.             break;
  30.         case CommandType.Hiccup:
  31.             ProcessHiccup(cmd.Arg);
  32.             break;
  33.         case CommandType.PipeTerm:
  34.             ProcessPipeTerm();
  35.             break;
  36.         case CommandType.PipeTermAck:
  37.             ProcessPipeTermAck();
  38.             break;
  39.         case CommandType.TermReq:
  40.             ProcessTermReq((Own)cmd.Arg);
  41.             break;
  42.         case CommandType.Term:
  43.             ProcessTerm((int)cmd.Arg);
  44.             break;
  45.         case CommandType.TermAck:
  46.             ProcessTermAck();
  47.             break;
  48.         case CommandType.Reap:
  49.             ProcessReap((SocketBase)cmd.Arg);
  50.             break;
  51.         case CommandType.Reaped:
  52.             ProcessReaped();
  53.             break;
  54.         default:
  55.             throw new ArgumentException();
  56.     }
  57. }

处理进程间通信协议

当创建进程间通信socket时,会调用ZObejct的RegisterEndpoint将socket对象加入到Context的使用inpoc协议的socket字段缓存中,而ZObject实际是调用Context的方法RegisterEndpoint,释放使用inpoc协议的socket和使用inpoc进行连接方式和RegisterEndpoint一样。

  1. protected void RegisterEndpoint(String addr, Ctx.Endpoint endpoint)
  2. {
  3.     //m_ctx是在ZObejct初始化是传进来的Context引用
  4.     m_ctx.RegisterEndpoint(addr, endpoint);
  5. }

多个IO线程

默认的IO线程数量是1个,当然也可以使用多个IO线程并发去处理,因此当创建监听对象或创建连接时则需要进行负载均衡,平分到多个IO线程去处理,切换IO线程也是在Context中实现的。

  1. protected IOThread ChooseIOThread(long affinity)
  2. {
  3.     return m_ctx.ChooseIOThread(affinity);
  4. }
  1. public IOThread ChooseIOThread(long affinity)
  2. {
  3.     //affinity表示哪些IO线程有资格,默认为0表示所有IO线程都可以处理。
  4.     if (m_ioThreads.Count == 0)
  5.         return null;
  6.     //  Find the I/O thread with minimum load.
  7.     int minLoad = -1;
  8.     IOThread selectedIOThread = null;
  9.     for (int i = 0; i != m_ioThreads.Count; i++)
  10.     {
  11.         if (affinity == 0 || (affinity & (1L << i)) > 0)
  12.         {
  13.             //获取IO线程socket载入次数
  14.             int load = m_ioThreads[i].Load;
  15.             //这里对IO线程进行负载均衡
  16.             if (selectedIOThread == null || load < minLoad)
  17.             {
  18.                 minLoad = load;
  19.                 selectedIOThread = m_ioThreads[i];
  20.             }
  21.         }
  22.     }
  23.     return selectedIOThread;
  24. }

总结

该篇介绍了Context和ZObject。NetMQ所有的socket对象创建,释放都离不开Context,由于Context内部对必要操作都加了锁,因此它是线程安全的。

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