Python3 实现 Redis 消息队列

概述废话不多说，我们先封装一个类。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.fr

废话不多说，我们先封装一个类。

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 = name    def publish(self,data):    	""" 发布 """        self.clIEnt.publish(self.queue_name,data)        return True    def subscribe(self):    	""" 订阅 """        pub = self.clIEnt.pubsub()        pub.subscribe(self.queue_name)        return pub    def run_subscribe(self):    	""" 启动订阅 """        pub = self.subscribe()        while True:            _,queue_name,message = pub.parse_response()            if _ == 'subscribe':                print('... 队列启动，开始接受消息 ...')                continue            data = {'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!'}

总结

以上是内存溢出为你收集整理的Python3 实现 Redis 消息队列全部内容，希望文章能够帮你解决Python3 实现 Redis 消息队列所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。