安裝操作庫

pip install stomp.py

?

基本寫法：

# -*-coding:utf-8-*- import stomp import timequeue_name = '/queue/SampleQueue' topic_name = '/topic/SampleTopic' listener_name = 'SampleListener'class SampleListener(object):def on_message(self, headers, message):print 'headers: %s' % headersprint 'message: %s' % message# 推送到隊列queue def send_to_queue(msg):conn = stomp.Connection10([('127.0.0.1',61613)])conn.start()conn.connect()conn.send(queue_name, msg)conn.disconnect()#推送到主題 def send_to_topic(msg):conn = stomp.Connection10([('127.0.0.1',61613)])conn.start()conn.connect()conn.send(topic_name, msg)conn.disconnect()##從隊列接收消息 def receive_from_queue():conn = stomp.Connection10([('127.0.0.1',61613)])conn.set_listener(listener_name, SampleListener())conn.start()conn.connect()conn.subscribe(queue_name)time.sleep(1) # secsconn.disconnect()##從主題接收消息 def receive_from_topic():conn = stomp.Connection10([('127.0.0.1',61613)])conn.set_listener(listener_name, SampleListener())conn.start()conn.connect()conn.subscribe(topic_name)while 1:send_to_topic('topic')time.sleep(3) # secsconn.disconnect()if __name__=='__main__':# send_to_queue('len 123')# receive_from_queue()receive_from_topic()

?

發送數據到隊列

發送端

import stomp import time# 通過調用stomp下的Connection10方法，創建連接，指定ip和端口 conn = stomp.Connection10([("47.dsadsa74.89", 61613)]) # 啟動，連接創建完成 conn.start() # 但是activemq是需要賬號密碼的，因此這里連接到指定用戶上面 conn.connect(username="admin", passcode="admin", wait=True)# 發送消息 for i in range(10):# 第一個參數：隊列的名稱# 第二個參數：消息的內容conn.send("queue1", f"message1:{i}")time.sleep(3)# 斷開連接 conn.disconnect()

接收端

import stompclass SampleListener(stomp.ConnectionListener):def on_message(self, headers, message):# headers是一個字典，里面很多內容，是一個字典，headers['destination']是對應的隊列名稱print(f"headers:{headers['destination']}, message:{message}")def on_error(self, headers, message):print(f"headers:{headers['destination']}, message:{message}")# 同樣創建一個連接 conn = stomp.Connection10([("47.9dsad.89", 61613)]) # 綁定監聽器，我們這里只有一個，所以名字什么的無所謂 conn.set_listener("", SampleListener()) # 啟動 conn.start() # 連接到指定用戶 conn.connect(username="admin", passcode="admin", wait=True) # 訂閱到指定的隊列，這個要和發送端發送的隊列保持一致 # 當發送端有消息過來時，消息的內容會自動傳遞到SampleListener類下的on_message的message參數里。headers則包含了一些額外信息，比如時間戳、隊列名等等 # 當然，如果有錯誤就會傳到on_error中，不過這里沒有定義，可以的話定義一下也是好的，參數和on_message是一樣的 conn.subscribe("queue1")# 讓程序不停下，不然程序立馬就結束了 while True:pass """ headers:/queue/queue1, message:message1:0 headers:/queue/queue1, message:message1:1 headers:/queue/queue1, message:message1:2 headers:/queue/queue1, message:message1:3 headers:/queue/queue1, message:message1:4 headers:/queue/queue1, message:message1:5 headers:/queue/queue1, message:message1:6 headers:/queue/queue1, message:message1:7 headers:/queue/queue1, message:message1:8 headers:/queue/queue1, message:message1:9 """

?

同時指定多個消費者

發送端

import stompconn = stomp.Connection10([("47.94dasd.89", 61613)]) conn.start() conn.connect(username="admin", passcode="admin", wait=True)for i in range(10):conn.send("queue1", f"message1:{i}")conn.send("queue2", f"message2:{i}")conn.disconnect()

接收端1

import stompclass SampleListener(stomp.ConnectionListener):def on_message(self, headers, message):print(f"headers:{headers['destination']}, message:{message}")def on_error(self, headers, message):print(f"headers:{headers['destination']}, message:{message}")conn = stomp.Connection10([("47.dasdas4.89", 61613)]) conn.set_listener("", SampleListener()) conn.start() conn.connect(username="admin", passcode="admin", wait=True) conn.subscribe("queue1")while True:pass

接收端2

import stompclass SampleListener(stomp.ConnectionListener):def on_message(self, headers, message):print(f"headers:{headers['destination']}, message:{message}")def on_error(self, headers, message):print(f"headers:{headers['destination']}, message:{message}")conn = stomp.Connection10([("47.9dasdsa89", 61613)]) conn.set_listener("", SampleListener()) conn.start() conn.connect(username="admin", passcode="admin", wait=True) conn.subscribe("queue1")while True:pass

?

參考：

https://www.cnblogs.com/traditional/p/11144123.html

總結

以上是生活随笔為你收集整理的Python连接ActiveMQ的操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。