Ubuntu 20.04下ROS与自研应用通过ZeroMQ互通信实现方案咨询

一、环境准备 #

• 确认已安装ROS Noetic（未安装的话先完成基础ROS安装流程）
• 安装ZeroMQ依赖：

  sudo apt update && sudo apt install libzmq3-dev python3-pyzmq
  

• 创建并初始化ROS工作空间（已有则跳过）：

  mkdir -p ~/zmq_ros_ws/src && cd ~/zmq_ros_ws/src
  catkin_init_workspace
  cd .. && catkin_make
  source devel/setup.bash
  

二、ROS端桥接节点实现 #

需要编写两个桥接节点：一个将ZeroMQ消息转为ROS话题发布，另一个将ROS话题消息转为ZeroMQ发送给自研应用。

1. ZeroMQ转ROS发布节点（Python示例） #

在ROS工作空间创建功能包：

cd ~/zmq_ros_ws/src
catkin_create_pkg zmq_ros_bridge rospy std_msgs
cd zmq_ros_bridge/scripts
touch zmq_to_ros.py && chmod +x zmq_to_ros.py

编辑zmq_to_ros.py内容：

#!/usr/bin/env python3
import rospy
import zmq
from std_msgs.msg import String

def zmq_to_ros():
    rospy.init_node('zmq_to_ros_bridge', anonymous=True)
    pub = rospy.Publisher('zmq_input_topic', String, queue_size=10)
    
    # 初始化ZeroMQ订阅端
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.connect("tcp://127.0.0.1:5555")
    socket.setsockopt_string(zmq.SUBSCRIBE, "")  # 订阅所有消息

    rate = rospy.Rate(10)
    while not rospy.is_shutdown():
        try:
            zmq_msg = socket.recv_string(flags=zmq.NOBLOCK)
            rospy.loginfo(f"Received from ZeroMQ: {zmq_msg}")
            ros_msg = String(data=zmq_msg)
            pub.publish(ros_msg)
        except zmq.Again:
            pass
        rate.sleep()

if __name__ == '__main__':
    try:
        zmq_to_ros()
    except rospy.ROSInterruptException:
        pass

2. ROS转ZeroMQ发布节点（Python示例） #

在同目录创建ros_to_zmq.py：

touch ros_to_zmq.py && chmod +x ros_to_zmq.py

编辑内容：

#!/usr/bin/env python3
import rospy
import zmq
from std_msgs.msg import String

socket = None

def ros_msg_callback(msg):
    try:
        socket.send_string(msg.data)
        rospy.loginfo(f"Sent to ZeroMQ: {msg.data}")
    except Exception as e:
        rospy.logerr(f"ZeroMQ send failed: {e}")

def ros_to_zmq():
    global socket
    rospy.init_node('ros_to_zmq_bridge', anonymous=True)
    rospy.Subscriber('ros_output_topic', String, ros_msg_callback)
    
    # 初始化ZeroMQ发布端
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.bind("tcp://127.0.0.1:5556")

    rospy.spin()

if __name__ == '__main__':
    try:
        ros_to_zmq()
    except rospy.ROSInterruptException:
        pass

3. 编译配置 #

修改package.xml，添加依赖声明：

<exec_depend>rospy</exec_depend>
<exec_depend>std_msgs</exec_depend>
<exec_depend>python3-pyzmq</exec_depend>

修改CMakeLists.txt，添加脚本安装规则：

catkin_install_python(PROGRAMS scripts/zmq_to_ros.py scripts/ros_to_zmq.py
  DESTINATION ${CATKIN_PACKAGE_BIN_DESTINATION}
)

编译工作空间：

cd ~/zmq_ros_ws && catkin_make && source devel/setup.bash

三、自研应用端ZeroMQ实现（Python示例） #

1. 向ROS发送消息的代码 #

import zmq
import time

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.connect("tcp://127.0.0.1:5555")

while True:
    msg = f"Custom app message at {time.time()}"
    socket.send_string(msg)
    print(f"Sent to ROS: {msg}")
    time.sleep(1)

2. 从ROS接收消息的代码 #

import zmq

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://127.0.0.1:5556")
socket.setsockopt_string(zmq.SUBSCRIBE, "")

while True:
    msg = socket.recv_string()
    print(f"Received from ROS: {msg}")

四、测试流程 #

1. 启动ROS核心：

   roscore
   

2. 启动两个桥接节点：

   rosrun zmq_ros_bridge zmq_to_ros.py
   rosrun zmq_ros_bridge ros_to_zmq.py
   

3. 运行自研应用发送脚本，通过ROS话题查看接收情况：

   rostopic echo /zmq_input_topic
   

4. 向ROS话题发布消息，查看自研应用接收脚本的输出：

   rostopic pub /ros_output_topic std_msgs/String "data: 'Hello from ROS'" -r 1
   

五、注意事项 #

• 端口号可按需修改，确保ROS端与自研应用端端口一致
• 若使用C++实现，需链接ZeroMQ库，核心API与Python逻辑一致（zmq::context_t、zmq::socket_t）
• 自定义消息类型需在ROS端定义.msg文件，自研应用端需配合JSON/Protobuf等完成序列化与反序列化

内容的提问来源于stack exchange，提问作者欧一文