廢話不多說，我們先封裝一個類。

redis_message_queue.py

#!/usr/bin/env python3 # -*- coding: UTF-8 -*-from redis import Redis, ConnectionPoolclass RMQ(object):def __init__(self, url, name):# self.client = Redis(host=url)pool = ConnectionPool.from_url(url=url, decode_responses=True)self.client = Redis(connection_pool=pool)self.queue_name = namedef publish(self, data):""" 發布 """self.client.publish(self.queue_name, data)return Truedef subscribe(self):""" 訂閱 """pub = self.client.pubsub()pub.subscribe(self.queue_name)return pubdef run_subscribe(self):""" 啟動訂閱 """pub = self.subscribe()while True:_, queue_name, message = pub.parse_response()if _ == 'subscribe':print('... 隊列啟動，開始接受消息 ...')continuedata = {'queue': queue_name, 'message': message}print(data)

publish.py

#!/usr/bin/env python3 # -*- coding: UTF-8 -*-from redis_message_queue import RMQrmq = RMQ(url='redis://127.0.0.1:6379/15', name='main')if __name__ == '__main__':print(rmq.publish('Nice!'))

執行結果：

(demo) MacBook:zhangyi$ python publish.py True (demo) MacBook: zhangyi$

subscribe.py

#!/usr/bin/env python3 # -*- coding: UTF-8 -*-from redis_message_queue import RMQrmq = RMQ(url='redis://127.0.0.1:6379/15', name='main')if __name__ == '__main__':rmq.run_subscribe()

執行結果：

(demo) MacBook:zhangyi$ python subscribe.py ... 隊列啟動，開始接受消息 ... {'queue': 'main', 'message': 'Nice!'}

參考文章：https://segmentfault.com/a/1190000016898228

總結

以上是生活随笔為你收集整理的Python3 实现 Redis 消息队列的全部內容，希望文章能夠幫你解決所遇到的問題。

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