Python的消息队列包SnakeMQ使用初探
发布时间：2020-12-16 20:33:42 所属栏目：Python 来源：网络整理
导读：一、关于snakemq的官方介绍 SnakeMQ的GitHub项目页：https://github.com/dsiroky/snakemq 1.纯python实现，跨平台 2.自动重连接 3.可靠发送--可配置的消息方式与消息超时方式 4.持久化/临时 两种队列 5.支持异步 -- poll() 6.symmetrical -- 单个TCP连接可用
|
一、关于snakemq的官方介绍 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,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,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,message):
print("message dropped",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)
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,ttl=60)
if self.messaging == None:
logger.error("listener:messaging is not initialized,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",4002)
niess.on_recv.add(received)
niess.start()
print("niess start...")
time.sleep(60)
niess.terminate()
print("niess end...")
（编辑：李大同） 【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容! |