使用 PHP 订阅者的 ZeroMQ 丢失消息

Question

我正在使用 zeromq 和 php 绑定连接到 Freeswitch（一个 VOIP 软件交换机）中的 zmq 模块。

简短：我正在失去活动。 长： Freeswitch 中的 zmq 模块是在 c++ 中作为发布者实现的。 我的PHP代码如下：

<?php
  $context = new ZMQContext();

  echo "connect to freeswitch zmq module...";
  $sub = new ZMQSocket($context, ZMQ::SOCKET_SUB);
  $sub->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE,"");
  $sub->connect("tcp://192.168.20.73:5556");
  $hwm = $sub->getSockOpt(ZMQ::SOCKOPT_HWM);
  echo "ok.hwm: $hwm\n";

  echo "looping\n";
  while(1) {
  $data = $sub->recv();
  $d = json_decode($sub->recv(),TRUE);
  $event = $d["Event-Name"];
  $date = $d["Event-Date-Local"];
  $ts = $d["Event-Date-Timestamp"];
  $msgnr = $d["ZMQ-Msg-Cnt"];
  echo "PHP: $date msg# $msgnr $ts received $event\n";
}
?>

ZMQ-Msg-Cnt 是我在 freeswitch 的 zmq 模块中构建的序列号。 我可以看到每第二条消息都丢失了。 tcpdump显示消息被zmq接收到了。

我已将 PHP 代码转换为 C，现在我能够接收每条消息。 C：

#include "zhelpers.h"
#include "cJSON.h"

int main (void)
{
    void *context = zmq_init (1);

    void *subscriber = zmq_socket (context, ZMQ_SUB);
    zmq_connect (subscriber, "tcp://192.168.20.73:5556");
    zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE, "", 0);

    while (1) {
        char *string = s_recv (subscriber);

        cJSON *root = cJSON_Parse(string);
        int msgcnt = cJSON_GetObjectItem(root,"ZMQ-Msg-Cnt")->valueint;

        printf("C: %s msg# %s %s received %s\n",
            cJSON_GetObjectItem(root,"Event-Date-Local")->valuestring,
            cJSON_GetObjectItem(root,"ZMQ-Msg-Cnt")->valuestring,
            cJSON_GetObjectItem(root,"Event-Date-Timestamp")->valuestring,
            cJSON_GetObjectItem(root,"Event-Name")->valuestring
        );
        cJSON_Delete(root);
        free (string);
    }

zmq_close (subscriber);
zmq_term (context);
return 0;
}

PHP 代码有什么问题吗？ PHP有什么技巧/必须做/提示吗？

提前致谢， 杰拉德·韦伯

Answer 1

您正在调用recv 函数两次，因此它正在加载一条消息，跳过对其的任何处理，然后加载第二条：

$data = $sub->recv();
// This is your first message, called in a blocking mode

$d = json_decode($sub->recv(),TRUE);
// and here's your second one, called in a non-blocking mode

将这两行改为一行：

$d = json_decode($sub->recv());