关闭Zeromq订户的正确方法是什么?

What is the proper way to shutdown a ZeroMQ Subscriber?

本文关键字：方法 是什么 Zeromq 关闭      更新时间：2023-10-16

我正在使用ros中的Zeromq中的PUB/SUB模型。

SUB -subscriber仅通过按 ctrl c c c 。

但是，每次我实际按下 ctrl c 时，它会停止以下错误：

^Cterminate called after throwing an instance of 'zmq::error_t'
  what():  Interrupted system call
Aborted (core dumped)

以下是代码段：

#include <ros/ros.h>
#include <zmq.hpp>
int main(int argc, char **argv)
{
  ros::init(argc, argv, "node_name", ros::init_options::AnonymousName);
  ros::NodeHandle nh;
  ros::Publisher pub;
  //  Prepare context and publisher
  zmq::context_t zmq_context(1);
  zmq::socket_t zmq_socket(zmq_context, ZMQ_SUB);
  zmq_socket.connect("tcp://192.168.1.20:9001");
  std::string TOPIC = "";
  zmq_socket.setsockopt(ZMQ_SUBSCRIBE, TOPIC.c_str(), TOPIC.length()); // allow all messages
  int timeout = 1000;  // Timeout to get out of the while loop since recv is blocking
  zmq_socket.setsockopt(ZMQ_RCVTIMEO, &timeout, sizeof(timeout));
  while (ros::ok())
  {
      zmq::message_t msg;
      int rc = zmq_socket.recv(&msg);
      if (rc)
      {
          //receive data and prepare it for publishing
          pub.publish(data);
          ros::spinOnce();
      }
  }
  // Clean up the socket and context here
  zmq_socket.close();
  zmq_context.close();
  return 0;
}

如何避免错误以正确关闭订户？

没有有关如何被困和处理Ctrl C的详细信息，我将始终添加（C 绑定的详细信息可能与版本有所不同）：

int main(int argc, char **argv)
{
    zmq_socket.connect(    "tcp://192.168.1.20:9001" );
    zmq_socket.setsockopt( ZMQ_LINGER, 0 );              // ALWAYS
    ...
    while( ROS::ok() )
    {
       ...
    }
    std::cout << "SIG:: will .close() after ROS::ok()-loop exit" << std::flush;
    zmq_socket.close();
    std::cout << "SIG:: will .term()  after a Socket()-instance .close()'d" << std::flush;
    zmq_context.close();
    std::cout << "SIG:: will return 0 after a Context()-instance .term()'d" << std::flush;
    return 0;
}

我意识到问题是由 int rc = zmq_socket.recv(&msg)引起的，因此我确实在改进之后做了 -

  1. 我使用try和catch块
  2. 按 @user3666197建议的设置ZMQ_LINGER

以下是代码段 - 

int linger = 0; // Proper shutdown ZeroMQ
zmq_socket.setsockopt(ZMQ_LINGER, &linger, sizeof(linger));
std::string socket_address = "tcp://192.168.1.20:9001";
zmq_socket.connect(socket_address.c_str());
ros::Duration duration(0.1); // in seconds (100 ms)
while (ros::ok())
{
    zmq::message_t msg;
    int rc = 0;
    try
    {
        rc = zmq_socket.recv(&msg);
    }
    catch (zmq::error_t& e)
    {
        ROS_DEBUG_STREAM("ZMQ Error. " << e.what());
    }
    if (rc)
    {
        unsigned char* byte_ptr = static_cast<unsigned char*>(msg.data());
        const int msg_length = msg.size();
        // receive data and prepare it for publishing
        pub.publish(data);
        ros::spinOnce();
    }
    else
    {
        ROS_DEBUG_STREAM("Pointcloud recv() returned 0");
        duration.sleep();
    }
}

谢谢

相关文章：