ZeroMQ (clrzmq4) 轮询问题答案

【问题标题】：ZeroMQ (clrzmq4) polling issueZeroMQ (clrzmq4) 轮询问题
【发布时间】：2016-08-06 03:23:58
【问题描述】：

我想要完成的是实现从两个套接字之一读取消息，无论它首先到达。据我了解，轮询 (zmq_poll) 是正确的做法（如 mspoller in guide 所示）。这里我提供小伪代码sn-p：

TimeSpan timeout = TimeSpan.FromMilliseconds(50);

using (var receiver1 = new ZSocket(ZContext.Current, ZSocketType.DEALER))
using (var receiver2 = new ZSocket(ZContext.Current, ZSocketType.PAIR))
{
    receiver1.Bind("tcp://someaddress");
    // Note that PAIR socket is inproc:
    receiver2.Connect("inproc://otheraddress");

    var poll = ZPollItem.CreateReceiver();

    ZError error;
    ZMessage msg;

    while (true)
    {
        if (receiver1.PollIn(poll, out msg, out error, timeout))
        {
            // ...
        }

        if (receiver2.PollIn(poll, out msg, out error, timeout))
        {
            // ...
        }
    }
}

如您所见，它实际上与mspoller in guide 中的完全相同。

在我的情况下，receiver2（PAIR 套接字）应该会收到大量消息。事实上，我创建了一个测试，其中发送给它的消息数量总是大于它能够接收的消息数量（至少在演示的实现中）。

我已经运行了 2 秒的测试，结果让我非常惊讶：

发送到receiver2 的消息数：180（“发送”是指将它们分发到另一个 PAIR 套接字，未在之前的 sn-p 中显示）；
receiver2 收到的消息数：21 ??? 2 秒内只有 21 条消息？？？每秒 10 条消息？？？

然后我尝试使用不同的timeout 值，我发现它会显着影响收到的消息数量。持续时间（2 秒）和发送的消息数（180）保持不变。结果是：

timeout 值为 200 毫秒 - 收到的消息数下降到 10（每秒 5 条）；
timeout 值为 10 毫秒 - 收到的消息数增加到 120 条（每秒 60 条）。

结果告诉我，轮询根本不起作用。如果轮询工作正常，据我了解机制，timeout 在这种情况下不应该有任何影响。无论我们将超时设置为 1 小时还是 5 毫秒 - 因为总是有消息要接收，所以无需等待，因此循环应该以相同的速度工作。

我的另一个大问题是，即使timeout 的值非常小，receiver2 也无法接收所有 180 条消息。我在这里努力实现每秒 100 条消息的接收率，尽管我选择了应该非常快的 ZeroMQ（基准提到的数字是每秒 600 万条消息）。

所以我的问题很明显：我在这里做错了吗？有没有更好的方法来实现轮询？

通过浏览 clrzmq4 代码，我注意到在枚举套接字 ZPollItems.cs, line 151 时也可以调用 pollIn 方法，但我在任何地方都没有找到任何示例！

这是正确的方法吗？有任何文档吗？

谢谢

【问题讨论】：

标签： zeromq

【解决方案1】：

我已经找到了问题/解决方案。而不是分别在每个套接字上使用PollIn 方法，我们应该在套接字数组上使用PollIn 方法。显然the example from the guide 是严重误导。下面是正确的做法：

TimeSpan timeout = TimeSpan.FromMilliseconds(50);

using (var receiver1 = new ZSocket(ZContext.Current, ZSocketType.DEALER))
using (var receiver2 = new ZSocket(ZContext.Current, ZSocketType.PAIR))
{
    receiver1.Bind("tcp://someaddress");
    receiver2.Connect("inproc://otheraddress");

    // We should "remember" the order of sockets within the array
    // because order of messages in the received array will correspond to it.
    ZSocket[] sockets = { receiver1, receiver2 };

    // Note that we should use two ZPollItem instances:
    ZPollItem[] pollItems = { ZPollItem.CreateReceiver(), ZPollItem.CreateReceiver() };

    ZError error;
    ZMessage[] msg;

    while (true)
    {
        if (sockets.PollIn(pollItems, out msg, out error, timeout))
        {
            if (msg[0] != null)
            {
                // The first message gotten from receiver1
            }

            if (msg[1] != null)
            {
                // The second message gotten from receiver2
            }
        }
    }
}

现在receiver2 每秒收到 15,000 条消息，无论timeout 的值如何，也无论receiver1 收到的消息数量如何。

更新：来自 clrzmq4 的人们已经确认 this issue，因此该示例可能很快会得到纠正。

【讨论】：