一、关于snakemq的官方介绍
SnakeMQ的GitHub项目页:https://github.com/dsiroky/snakemq
1.纯python实现,跨平台
2.自动重连接
3.可靠发送--可配置的消息方式与消息超时方式
4.持久化/临时 两种队列
5.支持异步 -- poll()
6.symmetrical -- 单个TCP连接可用于双工通讯
7.多数据库支持 -- SQLite、MongoDB……
8.brokerless - 类似ZeroMQ的实现原理
9.扩展模块:RPC, bandwidth throttling
以上都是官话,需要自己验证,动手封装了一下,感觉萌萌哒。
二、几个主要问题说明
1.支持自动重连,不需要自己动手写心跳逻辑,你只需要关注发送和接收就行
2.支持数据持久化,如果开始持久化,在重连之后会自动发送数据。
3.数据的接收,snakemq通过提供回调实现,你只需要写个接收方法添加到回调列表里去。
4.数据的发送,在此发送的都是bytes类型(二进制),因此需要转换。我在程序中测试的都是文本字符串,使用str.encode(‘utf-8')转换成bytes,接收时再转换回来。
5.术语解释,Connector:类似于socket的TcpClient,Lisenter:类似于socket的TcpServer,每个connector或者listener都一个一个ident标识,发送和接收数据时就知道是谁的数据了。
6.使用sqlite持久化时,需要修改源码,sqlite3.connect(filename,check_same_thread = False),用于解决多线程访问sqlite的问题。(会不会死锁?)
7.启动持久化时,如果重新连上,则会自动发送,保证可靠。
8.为了封装的需要,数据接收以后,我通过callback方式传送出去。
三、代码
说明代码中使用了自定义的日志模块
from common import nxlogger import snakemqlogger as logger
可替换成logging的。
回调类(callbacks.py):
# -*- coding:utf-8 -*- '''synchronized callback''' class Callback(object): def __init__(self): self.callbacks = [] def add(self, func): self.callbacks.append(func) def remove(self, func): self.callbacks.remove(func) def __call__(self, *args, **kwargs): for callback in self.callbacks: callback(*args, **kwargs)
Connector类(snakemqConnector.py):
# -*- coding:utf-8 -*- import threading import snakemq import snakemq.link import snakemq.packeter import snakemq.messaging import snakemq.message from snakemq.storage.sqlite import SqliteQueuesStorage from snakemq.message import FLAG_PERSISTENT from common.callbacks import Callback from common import nxlogger import snakemqlogger as logger class SnakemqConnector(threading.Thread): def __init__(self, snakemqident = None, remoteIp = "localhost", remotePort = 9090, persistent = False): super(SnakemqConnector,self).__init__() self.messaging = None self.link = None self.snakemqident = snakemqident self.pktr = None self.remoteIp = remoteIp self.remotePort = remotePort self.persistent = persistent self.on_recv = Callback() self._initConnector() def run(self): logger.info("connector start...") if self.link != None: self.link.loop() logger.info("connector end...") def terminate(self): logger.info("connetor terminating...") if self.link != None: self.link.stop() self.link.cleanup() logger.info("connetor terminated") def on_recv_message(self, conn, ident, message): try: self.on_recv(ident, message.data.decode('utf-8'))#dispatch received data except Exception as e: logger.error("connector recv:{0}".format(e)) print(e) '''send message to dest host named destIdent''' def sendMsg(self, destIdent, byteseq): msg = None if self.persistent: msg = snakemq.message.Message(byteseq, ttl=60, flags=FLAG_PERSISTENT) else: msg = snakemq.message.Message(byteseq, ttl=60) if self.messaging == None: logger.error("connector:messaging is not initialized, send message failed") return self.messaging.send_message(destIdent, msg) ''' ''' def _initConnector(self): try: self.link = snakemq.link.Link() self.link.add_connector((self.remoteIp, self.remotePort)) self.pktr = snakemq.packeter.Packeter(self.link) if self.persistent: storage = SqliteQueuesStorage("SnakemqStorage.db") self.messaging = snakemq.messaging.Messaging(self.snakemqident, "", self.pktr, storage) else: self.messaging = snakemq.messaging.Messaging(self.snakemqident, "", self.pktr) self.messaging.on_message_recv.add(self.on_recv_message) except Exception as e: logger.error("connector:{0}".format(e)) finally: logger.info("connector[{0}] loop ended...".format(self.snakemqident))
Listener类(snakemqListener.py):
# -*- coding:utf-8 -*- import threading import snakemq import snakemq.link import snakemq.packeter import snakemq.messaging import snakemq.message from common import nxlogger import snakemqlogger as logger from common.callbacks import Callback class SnakemqListener(threading.Thread): def __init__(self, snakemqident = None, ip = "localhost", port = 9090, persistent = False): super(SnakemqListener,self).__init__() self.messaging = None self.link = None self.pktr = None self.snakemqident = snakemqident self.ip = ip; self.port = port self.connectors = {} self.on_recv = Callback() self.persistent = persistent self._initlistener() ''' thread run ''' def run(self): logger.info("listener start...") if self.link != None: self.link.loop() logger.info("listener end...") ''' terminate snakemq listener thread ''' def terminate(self): logger.info("listener terminating...") if self.link != None: self.link.stop() self.link.cleanup() logger.info("listener terminated") ''' receive message from host named ident ''' def on_recv_message(self, conn, ident, message): try: self.on_recv(ident, message.data.decode('utf-8'))#dispatch received data self.sendMsg('bob','hello,{0}'.format(ident).encode('utf-8')) except Exception as e: logger.error("listener recv:{0}".format(e)) print(e) def on_drop_message(self, ident, message): print("message dropped", ident, message) logger.debug("listener:message dropped,ident:{0},message:{1}".format(ident, message)) '''client connect''' def on_connect(self, ident): logger.debug("listener:{0} connected".format(ident)) self.connectors[ident] = ident self.sendMsg(ident, "hello".encode('utf-8')) '''client disconnect''' def on_disconnect(self, ident): logger.debug("listener:{0} disconnected".format(ident)) if ident in self.connectors: self.connectors.pop(ident) ''' listen start loop ''' def _initlistener(self): try: self.link = snakemq.link.Link() self.link.add_listener((self.ip, self.port)) self.pktr = snakemq.packeter.Packeter(self.link) self.pktr.on_connect.add(self.on_connect) self.pktr.on_disconnect.add(self.on_disconnect) if self.persistent: storage = SqliteQueuesStorage("SnakemqStorage.db") self.messaging = snakemq.messaging.Messaging(self.snakemqident, "", self.pktr, storage) else: self.messaging = snakemq.messaging.Messaging(self.snakemqident, "", self.pktr) self.messaging.on_message_recv.add(self.on_recv_message) self.messaging.on_message_drop.add(self.on_drop_message) except Exception as e: logger.error("listener:{0}".format(e)) finally: logger.info("listener:loop ended...") '''send message to dest host named destIdent''' def sendMsg(self, destIdent, byteseq): msg = None if self.persistent: msg = snakemq.message.Message(byteseq, ttl=60, flags=FLAG_PERSISTENT) else: msg = snakemq.message.Message(byteseq, ttl=60) if self.messaging == None: logger.error("listener:messaging is not initialized, send message failed") return self.messaging.send_message(destIdent, msg)
测试代码connector(testSnakeConnector.py):
读取本地一个1M的文件,然后发送给listener,然后listener发回一个hello的信息。
from netComm.snakemq import snakemqConnector import time import sys import os def received(ident, data): print(data) if __name__ == "__main__": bob = snakemqConnector.SnakemqConnector('bob',"10.16.5.45",4002,True) bob.on_recv.add(received) bob.start() try: with open("testfile.txt",encoding='utf-8') as f: txt = f.read() for i in range(100): bob.sendMsg("niess",txt.encode('utf-8')) time.sleep(0.1) except Exception as e: print(e) time.sleep(5) bob.terminate() 测试代码listener(testSnakeListener.py): from netComm.snakemq import snakemqListener import time def received(ident, data): filename = "log/recFile{0}.txt".format(time.strftime('%S',time.localtime())) file = open(filename,'w') file.writelines(data) file.close() if __name__ == "__main__": niess = snakemqListener.SnakemqListener("niess","10.16.5.45",4002) niess.on_recv.add(received) niess.start() print("niess start...") time.sleep(60) niess.terminate() print("niess end...")
《魔兽世界》大逃杀!60人新游玩模式《强袭风暴》3月21日上线
暴雪近日发布了《魔兽世界》10.2.6 更新内容,新游玩模式《强袭风暴》即将于3月21 日在亚服上线,届时玩家将前往阿拉希高地展开一场 60 人大逃杀对战。
艾泽拉斯的冒险者已经征服了艾泽拉斯的大地及遥远的彼岸。他们在对抗世界上最致命的敌人时展现出过人的手腕,并且成功阻止终结宇宙等级的威胁。当他们在为即将于《魔兽世界》资料片《地心之战》中来袭的萨拉塔斯势力做战斗准备时,他们还需要在熟悉的阿拉希高地面对一个全新的敌人──那就是彼此。在《巨龙崛起》10.2.6 更新的《强袭风暴》中,玩家将会进入一个全新的海盗主题大逃杀式限时活动,其中包含极高的风险和史诗级的奖励。
《强袭风暴》不是普通的战场,作为一个独立于主游戏之外的活动,玩家可以用大逃杀的风格来体验《魔兽世界》,不分职业、不分装备(除了你在赛局中捡到的),光是技巧和战略的强弱之分就能决定出谁才是能坚持到最后的赢家。本次活动将会开放单人和双人模式,玩家在加入海盗主题的预赛大厅区域前,可以从强袭风暴角色画面新增好友。游玩游戏将可以累计名望轨迹,《巨龙崛起》和《魔兽世界:巫妖王之怒 经典版》的玩家都可以获得奖励。
更新动态
- 群星.1999-上华世纪国语总冠军4CD【上华】【WAV+CUE】
- 叶丽仪.1992-仙鹤神针电视剧原声音乐专辑【星传真】【WAV+CUE】
- 陈柏宇.2011-QUINQUENNIUM【SONY】【WAV+CUE】
- 《51首热门当红歌曲 十大顶级发烧女声 3CD》[WAV+CUE][1.6GB]
- 《赵传 深受摇滚迷青睐的台湾音乐人 约定》[WAV+CUE][320MB]
- 《庄心妍 实力唱将 发烧音乐精选集》[WAV+CUE][380MB]
- [音展纪念碟]2006年《TAA台湾第十六届国际HI-ENDHI-FI音响展》2cd[WAV]
- 中国音乐地图之听见西藏山歌牧歌酒歌2020[WAV分轨]
- 王宏伟2017-永远的红飘带[太平洋影音][WAV+CUE]
- 费翔.1997-NOW现在【第一中国】【WAV+CUE】
- 新格合唱团.1981-我唱你和2辑(滚石新格民歌系列)【新格】【WAV+CUE】
- 陈柏宇.2011-QUINQUENNIUM【SONY】【WAV+CUE】
- 雨果唱片-《金唛系列-蔷薇依韵琼瑶》WAV
- 陈百强《凝望(2012华纳LPCD45限量版)[DMI]》[WAV整轨]
- 模拟之声慢刻CD《徐小凤45周年》[德国母带][低速原抓WAV+CUE]