基于Redis做为消息中间件的代码封装,话不多说,直接上代码:
import json import threading try: from redis import StrictRedis, ConnectionPool except: import os os.system('pip install -i https://pypi.tuna.tsinghua.edu.cn/simple redis') from redis import StrictRedis, ConnectionPool def connect_redis(REDIS_HOST,REDIS_PORT,REDIS_DATABASE,REDIS_PWD): import socket #保持redis心跳的设置方式 pool = ConnectionPool(host=REDIS_HOST, port=REDIS_PORT,socket_keepalive=True, socket_keepalive_options={socket.TCP_KEEPIDLE: 60, socket.TCP_KEEPINTVL: 30, socket.TCP_KEEPCNT: 3} , db=REDIS_DATABASE, password=REDIS_PWD,decode_responses=True, encoding='UTF-8') gl_Redis = StrictRedis(connection_pool=pool,decode_responses=True, charset='UTF-8', encoding='UTF-8') try: gl_Redis.ping() msg = '【数据库:redis】 连接数据库成功...' print(msg) return gl_Redis except TimeoutError: msg = '【数据库:redis】 连接数据库失败...' print(msg) class RedisMQ(): def __init__(self,gl_Redis): self.coon=gl_Redis self.channels=[] self.TOPIC_handler = {'test':print} # 主题处理程序 def publish(self,channel,message,): '''发送消息,指定频道''' print(channel,message,) if isinstance(message,dict): message = json.dumps(message) try: return self.coon.publish(channel,message),'' except Exception as e: # print(str(e)) return False, str(e) def subscribe(self,channel): '''订阅单个频道''' pub=self.coon.pubsub() pub.subscribe(channel) pub.parse_response() self.channels.append(channel) return pub def psubscribe(self,channels): '''订阅多个频道''' pub=self.coon.pubsub() pub.psubscribe(*channels) pub.parse_response() self.channels+=channels return pub def close(self): self.conn.close() def unsubscribe(self,channel):#取消订阅 self.coon.pubsub().unsubscribe(channel) self.channels.remove(channel) def continue_subscribe(self,channel): '''持续监听单个频道,返回解析数据''' r=self.subscribe(channel) while True: if not (msg:=r.parse_response()): msg= {} elif msg[0] == 'subscribe': msg= {} else: msg=msg[-2:] yield msg def continue_psubscribe(self,channels): '''持续监听多个频道,返回解析数据''' r=self.psubscribe(channels) while True: if not (msg:=r.parse_response()): msg=[] elif msg[0] == 'psubscribe': msg=[] else: msg=msg[-2:] yield msg # def recv_command(self,obj,TOPIC,handler) -> None: # '''接收指令处理函数''' # while True: # # 解析数据-单次消费连接 # data = next(self.continue_subscribe(TOPIC)) # try: # handler(obj,data) # except Exception as e: # print(e) def recv_command(self,obj) -> None: '''接收指令处理函数''' channels = list(self.TOPIC_handler.keys()) if channels: while True: # 解析数据-单次消费连接 # data = next(self.continue_subscribe(TOPIC)) msg = next(self.continue_psubscribe(channels)) if msg: channel, data = msg handler = self.TOPIC_handler.get(channel,print) try: handler(obj,data) except Exception as e: print(e) def listen(self,obj,TOPIC_handler:dict) -> bool: try: for TOPIC,handler in TOPIC_handler.items(): print(f'【监听】频道:{TOPIC},消息监听程序启动!') # t3 = threading.Thread(target=self.recv_command,args=(obj,TOPIC,handler)) # 单次消费 t3 = threading.Thread(target=self.recv_command,args=(obj,)) # 单次消费 t3.start() return True except Exception as e: print(e) return False if __name__=="__main__": #订阅端示例: from sys_config.config import config # 换成你自己的配置信息 gl = config() gl_Redis = connect_redis(gl.REDIS_HOST,gl.REDIS_PORT,gl.REDIS_DATABASE,gl.REDIS_PWD) r=RedisMQ(gl_Redis) r.recv_command(None) # chat=r.subscribe('chat') #单条订阅示例 # chat = r.psubscribe(['para_data_1:1:3','device_login_1:1:3']) #多条订阅示例 # print('开始监听') # while True: # print('监听中') # # 解析数据 # msg=chat.parse_response()#block=False, timeout=60 参数解释,非阻塞模式,60秒无数据返回None # print(msg) # channel, data = msg[-2:] # print('频道:',channel) # print('数据:',data) # # 多条订阅数据解析示例 # chat = r.continue_psubscribe([]) # while True: # data=next(chat) #channel, # print('频道:', channel) # print('数据:', data) # # # 发布示例 # while True: # m = input("请输入你要发布的频道:") # n = input("请输入你要发布的消息:") # r.publish(m, n) # m = 'command_all' # n= '30 31 30 33 30 31 46 34 30 30 30 38 30 34 30 32'.replace(' ','') # n= '01 03 01 F4 00 08 04 02'.replace(' ','') # n= '03 03 00 00 00 01 85 E8'.replace(' ','') # n= '04 03 00 00 00 01 84 5F'.replace(' ','') # n= '05 03 00 00 00 01 85 8E'.replace(' ','') # n= '06 03 00 00 00 01 85 BD'.replace(' ','') # n= '07 03 00 00 00 02 C4 6D'.replace(' ','') # r.publish(m,n) # m = 'command_device' # # n = json.dumps({'device_id':"863100062008572","command":"010301F400080402"}) # n = json.dumps({'device_id':"863100061996678","command":"010301F400080402"}) # r.publish(m,n) # print(r.publish('hello',{}))