消息队列NetMQ 原理分析1-Context和ZObject
2017-01-15 17:37
543 查看
消息队列NetMQ 原理分析1-Context和ZObject
前言
介绍
目的
Context
初始化Context
创建SocketBase
释放SocketBase
缓存进程内通信Socket
ZObject
处理进程间通信协议
多个IO线程
总结
zeromq的英文文档
NetMQ的英文文档
消息队列NetMQ 原理分析1-Context和ZObject
消息队列NetMQ 原理分析2-IO线程和完成端口
消息队列NetMQ 原理分析3-命令产生/处理、创建Socket和回收线程
消息队列NetMQ 原理分析4-Session、Option和Pipe
消息队列NetMQ 原理分析5-Engine
消息队列NetMQ 原理分析6-TCP和Inpoc实现
消息队列NetMQ 原理分析7-Device
消息队列NetMQ 原理分析8-不同类型的Socket
消息队列NetMQ 原理分析9-实战
其余1024个slot保存的是socket对象的Mailbox,当socket还没使用是,slots保存的是null,占个位置,同时m_emptySlots。
Context会用一个字典管理当前使用inpoc的socket。
当inpoc的socket进行绑定时会加入到字典缓存中。释放时会从字典缓存中移除。当使用inpoc协议连接时,增加当前绑定inpoc地址的连接数。
NetMQ中一共定义了一下的命令类型
根据不同命令类型进行处理,处理方式由具体的Socket子类去重载。
前言
介绍
目的
Context
初始化Context
创建SocketBase
释放SocketBase
缓存进程内通信Socket
ZObject
处理进程间通信协议
多个IO线程
总结
前言
介绍
[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的源码进行学习并分析理解,因此写下该系列文章,本系列文章暂定编写计划如下:消息队列NetMQ 原理分析1-Context和ZObject
消息队列NetMQ 原理分析2-IO线程和完成端口
消息队列NetMQ 原理分析3-命令产生/处理、创建Socket和回收线程
消息队列NetMQ 原理分析4-Session、Option和Pipe
消息队列NetMQ 原理分析5-Engine
消息队列NetMQ 原理分析6-TCP和Inpoc实现
消息队列NetMQ 原理分析7-Device
消息队列NetMQ 原理分析8-不同类型的Socket
消息队列NetMQ 原理分析9-实战
Context
NetMQ有一个Context对象,用于初始化并保存当前NetMQ底层的对象状态,如IO线程、回收线程、进程间传输节点字典、插槽m_slots(用于保存IO对象,回收对象和socket对象的Mailbox)、初始化但未用到的Socket对象指针数组以及当前Mailbox(用于接收终止信号)等。初始化Context
当创建第一个Socket对象时会初始化IO线程,回收线程以及工作线程。默认Socket数量1024个,IO线程1个,回收线程1个。m_slots = new Mailbox[m_slotCount];//m_soltCount = 1 + 1 + 1024
m_slots[0]保存的是Context的Mailbox。
m_slots[TermTid] = m_termMailbox;//用于当前Context接收终止信号
m_slots[1]保存的是回收对象的Mailbox,保存完毕后就会启动回收对象轮询线程。
m_reaper = new Reaper(this, ReaperTid);//ReaperTid = 1 m_slots[ReaperTid] = m_reaper.Mailbox; m_reaper.Start();
m_slots[2]保存的是IO线程对象的Mailbox。
for (int i = 2; i != ios/*ios = 1,默认用1个io线程*/ + 2; i++) { IOThread ioThread = new IOThread(this, i); m_ioThreads.Add(ioThread); m_slots[i] = ioThread.Mailbox; ioThread.Start(); }
其余1024个slot保存的是socket对象的Mailbox,当socket还没使用是,slots保存的是null,占个位置,同时m_emptySlots。
//m_soltCount = 1 + 1 + 1024 for (int i = (int)m_slotCount - 1; >= (int)ios + 2; i--) { m_emptySlots.Push(i); m_slots[i] = null; }
创建SocketBase
无论是什么类型的Socket全都是在Context中进行创建或释放的。NetMQ中不同Socket都继承自SocketBase,在Context未中止且Socket未满时,会从m_emptySlots栈中Pop出一个未使用的指针。若创建失败,则重新加回到栈中,否则更新当前使用的Socket的集合加入该Socket并更新
m_slots的Mailbox
//slot是当前socket在s_slots中的位置,也用于生成SocketBase的`ThreadId` int slot = m_emptySlots.Pop(); // sid是生成并递增的唯一的socket ID,用于SocketBase创建MailBox命名用,并无实际其他作用。 int sid = Interlocked.Increment(ref s_maxSocketId); s = SocketBase.Create(type, this, slot, sid); if (s == null) { m_emptySlots.Push(slot); return null; } m_sockets.Add(s); m_slots[slot] = s.Mailbox;
释放SocketBase
当Reaper要释放某个SocketBase时,最终会调用Context的
DestroySocket方法。
tid = socket.ThreadId; //重新加入到可用socket栈中 m_emptySlots.Push(tid); //关闭连接 m_slots[tid].Close(); //清空引用 m_slots[tid] = null; // 从当前使用socket集合移除 m_sockets.Remove(socket); //若当前接收到中止信号且当前socket全部已释放时停止回收线程 if (m_terminating && m_sockets.Count == 0) 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中一共定义了一下的命令类型
public enum CommandType { // 发送给IO线程表示当前对象需要停止 Stop, // 发送给IO线程表示当前对象需要注册到IO线程中 Plug, // 将创建的对象Session的加入到当前Socket的所属集合中 Own, // 附加engine到Session中 Attach, // 建立session到Socket之间的管道,在握手之前调用inc_seqnum. Bind, // 通过写管道发送通知给读管道多少信息可读 ActivateRead, // 通过读管道发送通知给写读管道多少信息可写 ActivateWrite, // 创建一个新的管道后通过读管道发送给写管道 // 参数是管道类型,然而,他的目的地是私有的,因此我们必须用void指针, however, Hiccup, // 通过读管道发送到写管道告诉他中止所有管道 PipeTerm, // 写管道对PipeTerm命令响应 PipeTermAck, // 通过IO对象发送给socket请求终端IO对象 TermReq, // 通过socket发送给IO对象他自己开始关闭 Term, // 通过IO对象发送给socket让它知道已经关闭 TermAck, // 将关闭套接字的所有权转移给回收线程. Reap, // 关闭套接字通知回收线程他已经释放 Reaped, // 当所有socket都被释放通过回收线程发送给 term 线程 Done }
根据不同命令类型进行处理,处理方式由具体的Socket子类去重载。
public void ProcessCommand(Command cmd) { switch (cmd.CommandType) { case CommandType.ActivateRead: ProcessActivateRead(); break; case CommandType.ActivateWrite: ProcessActivateWrite((long)cmd.Arg); break; case CommandType.Stop: ProcessStop(); break; case CommandType.Plug: ProcessPlug(); ProcessSeqnum(); break; case CommandType.Own: ProcessOwn((Own)cmd.Arg); ProcessSeqnum(); break; case CommandType.Attach: ProcessAttach((IEngine)cmd.Arg); ProcessSeqnum(); break; case CommandType.Bind: ProcessBind((Pipe)cmd.Arg); ProcessSeqnum(); break; case CommandType.Hiccup: ProcessHiccup(cmd.Arg); break; case CommandType.PipeTerm: ProcessPipeTerm(); break; case CommandType.PipeTermAck: ProcessPipeTermAck(); break; case CommandType.TermReq: ProcessTermReq((Own)cmd.Arg); break; case CommandType.Term: ProcessTerm((int)cmd.Arg); break; case CommandType.TermAck: ProcessTermAck(); break; case CommandType.Reap: ProcessReap((SocketBase)cmd.Arg); break; case CommandType.Reaped: ProcessReaped(); break; default: throw new ArgumentException(); } }
处理进程间通信协议
当创建进程间通信socket时,会调用ZObejct的RegisterEndpoint将socket对象加入到Context的使用inpoc协议的socket字段缓存中,而ZObject实际是调用Context的方法
RegisterEndpoint,释放使用inpoc协议的socket和使用inpoc进行连接方式和
RegisterEndpoint一样。
protected void RegisterEndpoint(String addr, Ctx.Endpoint endpoint) { //m_ctx是在ZObejct初始化是传进来的Context引用 m_ctx.RegisterEndpoint(addr, endpoint); }
多个IO线程
默认的IO线程数量是1个,当然也可以使用多个IO线程并发去处理,因此当创建监听对象或创建连接时则需要进行负载均衡,平分到多个IO线程去处理,切换IO线程也是在Context中实现的。protected IOThread ChooseIOThread(long affinity) { return m_ctx.ChooseIOThread(affinity); }
public IOThread ChooseIOThread(long affinity) { //affinity表示哪些IO线程有资格,默认为0表示所有IO线程都可以处理。 if (m_ioThreads.Count == 0) return null; // Find the I/O thread with minimum load. int minLoad = -1; IOThread selectedIOThread = null; for (int i = 0; i != m_ioThreads.Count; i++) { if (affinity == 0 || (affinity & (1L << i)) > 0) { //获取IO线程socket载入次数 int load = m_ioThreads[i].Load; //这里对IO线程进行负载均衡 if (selectedIOThread == null || load < minLoad) { minLoad = load; selectedIOThread = m_ioThreads[i]; } } } return selectedIOThread; }
总结
该篇介绍了Context和ZObject。NetMQ所有的socket对象创建,释放都离不开Context,由于Context内部对必要操作都加了锁,因此它是线程安全的。相关文章推荐
- 消息队列NetMQ 原理分析3-命令产生/处理和回收线程
- 消息队列NetMQ 原理分析5-StreamEngine、Encord和Decord
- 消息队列NetMQ 原理分析2-IO线程和完成端口
- 消息队列NetMQ 原理分析4-Socket、Session、Option和Pipe
- 简要分析Ogre渲染队列的实现原理
- Linux消息队列原理与应用
- 简要分析Ogre渲染队列的实现原理(一)
- 简要分析Ogre的渲染队列实现原理(二)
- 消息队列(MQ)技术的介绍和原理
- 简要分析Ogre渲染队列的实现原理
- 简要分析Ogre渲染队列的实现原理(转)
- WM里蓝牙AVRCP的消息队列研究分析(研究ing,笔记)
- 等待队列的原理与源码分析
- [转]简要分析Ogre渲染队列的实现原理(一)
- [转]简要分析Ogre的渲染队列实现原理(二)
- ipcs 命令 分析消息队列、共享内存和信号量
- 消息队列技术的介绍和原理(MQ)
- 链式队列的c++代码和原理以及编译分析
- 简要分析Ogre渲染队列的实现原理
- 深入理解Android消息队列原理篇