在ros里面，是有專門的工具幫忙做這一步的，但是ros2里面還沒有，不過我看論壇上大家更加推薦使用代碼的形式做數(shù)據(jù)傳輸。

我使用的是python的paho這個(gè)包，首先需要安裝

pip install paho

我這里貼兩個(gè)代碼，分別是publisher和subscriber，也就是發(fā)布者和訂閱者。

1. publisher

import time
import paho.mqtt.client as mqtt




class Publisher:
    def __init__(self, host="127.0.0.1", port=1883, topic="test_channel"):
        self.host = host
        self.port = port
        self.topic = topic
        self.client = mqtt.Client()


        self.client.on_connect = self.on_connect
        self.client.on_publish = self.on_publish


    def on_connect(self, client, userdata, flags, rc):
       print("Connected with result code "+str(rc))


    def on_publish(self, client, userdata, mid):
        print("Message Published ...")


    def start(self, msg="Hello MQTT", times=10, delay=1):
        self.client.connect(self.host, self.port, 60)
        self.client.loop_start()


        for i in range(times):
            time.sleep(delay)
            self.client.publish(self.topic, f"{msg} {i}")




if __name__ == "__main__":
    publisher = Publisher()
    publisher.start()

2. subscriber

import paho.mqtt.client as mqtt




class Subscriber:
    def __init__(self, host="127.0.0.1", port=1883, topic="test_channel"):
        self.host = host
        self.port = port
        self.topic = topic
        self.msg_count = 0
        self.client = mqtt.Client()


        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message


    def on_connect(self, client, userdata, flags, rc):
        print("Connected with result code "+str(rc))
        self.client.subscribe(self.topic)


    def on_message(self, client, userdata, msg):
        self.msg_count += 1
        print(f"Message {self.msg_count}: {msg.topic} {str(msg.payload)}")


    def start(self):
        self.client.connect(self.host, self.port, 60)
        self.client.loop_forever()




if __name__ == "__main__":
    subscriber = Subscriber()
    subscriber.start()

可以在跟mosquitto所在的同一臺機(jī)器上運(yùn)行上面兩個(gè)腳本，否則就要修改代碼中的host為mosquitto實(shí)際的IP地址，還要確保網(wǎng)絡(luò)沒有限制。

測試的時(shí)候，要先運(yùn)行subscriber，然后再運(yùn)行publisher，否則subscriber很可能接受不到數(shù)據(jù)。