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

2022-11-25 13:04:05

前言

介绍

[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的源码进行学习并分析理解,因此写下该系列文章,本系列文章暂定编写计划如下：

  1. 消息队列NetMQ 原理分析1-Context和ZObject
  2. 消息队列NetMQ 原理分析2-IO线程和完成端口
  3. 消息队列NetMQ 原理分析3-命令产生/处理和回收线程
  4. 消息队列NetMQ 原理分析4-Session和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个。

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 原理分析1-Context和ZObject

微信扫一扫二维码关注订阅号杰哥技术分享

本文地址：https://www.cnblogs.com/Jack-Blog/p/6287458.html

作者博客：杰哥很忙

欢迎转载，请在明显位置给出出处及链接)

上一篇：PHP消息队列用法实例分析


下一篇：redis作为消息队列的原理