使用 ZeroMQ 在 C 中接收多部分消息答案

【问题标题】：Receive multipart messages in C with ZeroMQ使用 ZeroMQ 在 C 中接收多部分消息
【发布时间】：2016-07-22 19:54:22
【问题描述】：

我正在为将使用 ZeroMQ 套接字传输数据并发布多部分消息的设备编写数据记录系统。

我的程序需要用 C 编写。

我确实设法编写了一个完全可以正常工作的 Python 脚本。这看起来像这样：

import zmq
import time
import numpy as np

_HEARTBEAT_PREFIX =     b'ray.heartbeat\0'  
_RAW_DATA_PREFIX =      b'ray.rawdata\0'

context = zmq.Context.instance()

time.sleep(5)

socket = context.socket(zmq.SUB)
socket.setsockopt(zmq.LINGER, 0)
socket.setsockopt(zmq.SUBSCRIBE, b'')
time.sleep(5)
socket.connect("tcp://localhost:50290")
time.sleep(5)
go = True
while go:

    if socket.poll(1000):

        zframes = socket.recv_multipart(flags=zmq.NOBLOCK, copy=False)

        if (zframes[0].buffer.tobytes() == _HEARTBEAT_PREFIX):
            print ('Heartbeart')
        elif (zframes[0].buffer.tobytes() == _RAW_DATA_PREFIX):
            dataToWrite = np.frombuffer(zframes[1].buffer, dtype='i2')
            print ('Raw data: ')
            # and so on ...

这是我第一次使用 ZMQ，尝试将此 Python 代码移植到 C 是一个真正的挑战。我现在写的 C 代码可以运行，但是不能正常工作：

#include "zhelpers.h"

int main (int argc, char *argv [])
{

  void *context   = zmq_ctx_new();
  void *subscriber = zmq_socket(context,ZMQ_SUB);

  int rc = zmq_connect(subscriber,"tcp://localhost:50290");
  assert(rc == 0);
  rc = zmq_setsockopt(subscriber,ZMQ_SUBSCRIBE, "", 0);
  assert(rc == 0); 

  while (1) 
  {
    zmq_msg_t message;
    zmq_msg_init (&message);
    zmq_msg_recv (&message, subscriber,0);//, ZMQ_NOBLOCK);
    //  Process the message frame


    int size = zmq_msg_size(&message);
    char *string = malloc(size + 1);
    memcpy(string, zmq_msg_data(&message), size);


    zmq_msg_close (&message);

    string[size] = 0;
    printf("Message[%d]: %s\n", size, string);

    if (!zmq_msg_more (&message))
        break;      //  Last message frame
  }
  return 0;
}

然而，此代码将提供传入数据的正确大小，但在显示数据时，它显示为不可读的字符（符号等..），就像在 Python 中所做的那样，我需要将传入数据获取为一个字符串，以便进一步处理它。

【问题讨论】：

C 代码似乎没有检查帧类型（心跳/原始数据）。第一帧似乎是类型，如果类型是原始数据，那么下一帧包含可能是可打印的内容。

标签： c sockets zeromq

【解决方案1】：

我发现 Python 中的 numpy 数组被转换为包含 int 的字符串数组，它们的值很难为 C 程序解释。

通过处理接收到的数据，我能够提取正确的值：

#include "zhelpers.h"

int main (int argc, char *argv [])
{

  int i = 0;

  void *context   = zmq_ctx_new();
  void *subscriber = zmq_socket(context,ZMQ_SUB);

  int rc = zmq_connect(subscriber,"tcp://localhost:50290");
  assert(rc == 0);

  char* filter = "ray.";

  rc = zmq_setsockopt(subscriber,ZMQ_SUBSCRIBE, filter, sizeof(filter));
  assert(rc == 0); 
  int nextFrame = 0;

  while (1) 
  {

    zmq_msg_t message;
    zmq_msg_init (&message);
    zmq_msg_recv (&message, subscriber,0);//, ZMQ_NOBLOCK);

    //  Process the message frame

    int size = zmq_msg_size(&message);
    char *string = malloc(size + 1);
    memcpy(string, zmq_msg_data(&message), size);

    if (nextFrame)
    {

      zmq_msg_close (&message);
      printf("Raw data: "); 
      string[size] = 0;
      //printf("Data: %s\n", string);
      for ( i = 0; i < size; i+=2)
      {

        printf("%d ", (int16_t) ((string[i] & 0xff) +((string[i+1] & 0xff)<<8)));

      }


    }
      if (strcmp(string, "ray.rawdata") != 0)
      {
        nextFrame = 0;
        break;
      }
      else 
      {
        nextFrame = 1;

        zmq_msg_close (&message);

        string[size] = 0;
        printf("Prefix: %s\n", string);

        if (!zmq_msg_more (&message))
            break;      //  Last message frame
      }

  }}
  return 0;
}

请注意，这只有在从 Python 接收 numpy 数组时才会出现问题 - 只要它是心跳字符串或其他什么，就没有问题。

【讨论】：