c++ - ZMQ : Sending custom CPP object over the ZMQ queue

Question

我有一个名为 GenericMessage 的类，显示在下面的第一个代码片段中(在 GenericMessage.hxx 中定义)。

我有一个名为 TestFE.cpp 的 .cpp 文件(参见下面的第二个代码片段)，它试图通过 ZMQ 队列发送类 GenericMessage 的实例(另请参见下面的第四个代码片段 - ZmqHandler.hxx)。 TesfFE.cpp 通过包含 ZmqHandler.hxx 在此处实现 ZMQ 推送模式。

我还有另一个名为 TestBE.cpp 的 .cpp 文件(请参阅下面的第三个代码片段)，它通过 ZMQ 队列接收上述 GenericMessage 实例。 TestBE.cpp 在此处实现 ZMQ 拉模式以通过 ZMQ 队列检索 GenericMessage 实例。

在 TestFE.cpp 中，我使用标准的 memcpy 函数将 GenericMessage 对象转换为 ZMQ 队列可以接受的形式。在 TestBE.cpp 的第 21 行(在评论中的第三个代码片段中标记)，我得到了一个段错误，因为它看起来 memcpy 在发送方(即 TestFE.cpp)上无法正常工作。执行 TestBE 时收到以下消息。我还在下面提供了 gdb 回溯。你能告诉我这里出了什么问题吗？为什么您认为 memcpy 无法将我的 GenericMessage 对象正确复制为 ZMQ message_t 格式？还是您认为问题出在其他地方？如有任何意见，我们将不胜感激。

错误信息

  $ ./TestBE
  Connecting to FE...
  RECEIVED: 1
  Segmentation fault (core dumped)

GDB 回溯

  (gdb) r
   Starting program: /home/holb/HOLB_DESIGN/ZMQ/WORK1/TestBE 
  [Thread debugging using libthread_db enabled]
  Using host libthread_db library "/lib/i386-linux-gnu/libthread_db.so.1".
  [New Thread 0xb7c84b40 (LWP 4252)]
  [New Thread 0xb7483b40 (LWP 4253)]
  Connecting to FE...
  RECEIVED: 1

  Program received signal SIGSEGV, Segmentation fault.
  0xb7f371cc in std::basic_string<char, std::char_traits<char>, std::allocator<char>       >::basic_string(std::string const&) ()
  from /usr/lib/i386-linux-gnu/libstdc++.so.6
  (gdb) bt
  #0  0xb7f371cc in std::basic_string<char, std::char_traits<char>,      std::allocator<char> >::basic_string(std::string const&) ()
  from /usr/lib/i386-linux-gnu/libstdc++.so.6
  #1  0x08049621 in GenericMessage<std::string>::getData (this=0xbffff06c)
  at GenericMessage.hxx:18
  #2  0x08049075 in main () at TestBE.cxx:21
 (gdb) 

代码片段 1 (GenericMessage.hxx) ＃包括 ＃包括 #包括

  template <class T>
  class GenericMessage {
  public:

         GenericMessage(int id, T msg): 
         beId(id), 
         data(msg)
         {}

         ~GenericMessage(){}

         T getData()
         {
           //LINE 18 is the following line!
           return data;
         }

          std::string toString()
          {
              std::ostringstream ss;
              ss << getBeId();
              std::string ret =  ss.str();

              return ret;
          }

          void setBeId(int id)
          {
              beId = id;
          }

          int getBeId()
          {
             return beId;
          }
 private:
        int beId;
        T data;
 };

代码片段 2(TestFE.cxx ==> 发件人) #include "ZmqHandler.hxx" //ZmqHandler.hxx 的内容参见底部的第 4 个片段

 int main ()
 {
     ZmqHandler<std::string> zmqHandler;
     int counter = 1;

     while(1)
     {  
        std::string data = "Hello there!\0";
        GenericMessage<std::string> msg(counter, data);
        zmqHandler.sendToBE(&msg);
        counter++;
        sleep(1);
     }

     return 0;
 }

代码片段 3(TestBE.cxx ==> 接收器)

  #include "zmq.hpp"
  #include "GenericMessage.hxx"
  #include <string>
  #include <iostream>

  int main ()
  {
      //  Prepare our context and socket
      zmq::context_t context (1);
      zmq::socket_t socket (context, ZMQ_PULL);

     std::cout << "Connecting to FE..." << std::endl;
     socket.connect ("tcp://localhost:5555");

    while(1){
        zmq::message_t reply;
            socket.recv (&reply);
            GenericMessage<std::string> *msg = (GenericMessage<std::string>*)(reply.data());                
            std::cout << "RECEIVED: " << msg->toString() << std::endl;

           /* *********************************  */
           /*  SEGMENTATION FAULT HAPPENS HERE   */
           /* The member "data" in class GenericMessage cannot be received while the  member "id" in the previous line can be received. */
           std::cout << "DATA: " << ((std::string)msg->getData())  << std::endl;
           /* ********************************** */
     }

     return 0;
 }

代码片段 4 (ZMQHandler.hxx)

  #include "zmq.hpp"  
  #include "GenericMessage.hxx"
  #include <pthread.h>
  #include <unistd.h>
  #include <cassert>

  template <class T>
  class ZmqHandler {
  public:
        ZmqHandler():
    mContext(1),
    mOutbHandlerSocket(mContext, ZMQ_PUSH)
        {    
          mOutbHandlerSocket.bind ("tcp://*:5555");       
        }

       ~ZmqHandler() {}

       void *sendToBE(GenericMessage<T> *theMsg)
       {
         //  Place the new request to the zmq queue for BE consumption
         zmq::message_t msgToSend(sizeof(*theMsg));

         memcpy ( msgToSend.data(), ((GenericMessage<T>*)theMsg), sizeof(*  ((GenericMessage<T>*)theMsg)));

         mOutbHandlerSocket.send(msgToSend);

         std::cout << "SENT request: [" << theMsg->toString() << "]" << std::endl;

         return (NULL);
        }   

   private:  
        zmq::context_t mContext;
        zmq::socket_t mOutbHandlerSocket; 

};

Answer 1

我开始看到问题了。这是您发送完整的“结构”，其中包含一个具有指针的成员变量(std::string)。你不能这样做，因为指针只在创建它们的程序中有效。

你必须 serialize发送之前的结构，然后在接收端反序列化。

您可以使用 Boost serialization 等库为此，或 Google protocol buffers ，或任何其他数量的库。