python消息队列框架_python消息队列snakemq使⽤总结Python 消息队列snakemq总结
最近学习消息总线zeromq，在⽹上搜了python实现的消息总线模块，意外发现有个消息队列snakemq，于是拿来研究⼀下，感觉还是很不错的，⼊⼿简单使⽤也简单(⽐ice强多了)，就是资料太少了，只能⾃⼰抠。
⼀、关于snakemq的官⽅介绍
1纯python实现，跨平台
2⾃动重连接
3可靠发送--可配置的消息⽅式与消息超时⽅式
4持久化/临时 两种队列
5⽀持异步 -- poll()
6symmetrical -- 单个TCP连接可⽤于双⼯通讯
7多数据库⽀持 -- SQLite、MongoDB……
8brokerless - 类似ZeroMQ的实现原理
9扩展模块：RPC, bandwidth throttling
以上都是官话，需要⾃⼰验证，动⼿封装了⼀下，感觉萌萌哒。
⼆、⼏个主要问题说明
1⽀持⾃动重连，不需要⾃⼰动⼿写⼼跳逻辑，你只需要关注发送和接收就⾏
2⽀持数据持久化，如果开始持久化，在重连之后会⾃动发送数据。
3数据的接收，snakemq通过提供回调实现，你只需要写个接收⽅法添加到回调列表⾥去。
4数据的发送，在此发送的都是bytes类型(⼆进制)，因此需要转换。我在程序中测试的都是⽂本字符串，使⽤de(‘utf-8’)转换成bytes，接收时再转换回来。
5术语解释，Connector：类似于socket的TcpClient，Lisenter：类似于socket的TcpServer，每个connector或者listener都⼀个⼀个ident标识，发送和接收数据时就知道是谁的数据了。
6使⽤sqlite持久化时，需要修改源码，t(filename,check_same_thread = False)，⽤于
解决多线程访问sqlite的问题。(会不会死锁？)
7启动持久化时，如果重新连上，则会⾃动发送，保证可靠。
8为了封装的需要，数据接收以后，我通过callback⽅式传送出去。
三、代码
说明代码中使⽤了⾃定义的⽇志模块
from common import nxlogger
import snakemqlogger as logger
可替换成logging的。
回调类(callbacks.py)：
# -*- coding:utf-8 -*-
#*********************************************************#
# @@FileName: callbacks.py
# @@Author:
Shelwinnee
# @@Create Date: 2016-04-14 22:24:14
# @@Modify Date: 2016-04-14 22:24:34
# @@Description: simple synchronized
callbacks helper
#*********************************************************#
'''synchronized callback'''
class Callback(object):
def __init__(self):
self.callbacks = []
def add(self, func):
self.callbacks.append(func)
def remove(self, func):
ve(func)
def __call__(self, *args, **kwargs):
for callback in self.callbacks:
callback(*args, **kwargs)
Connector类(snakemqConnector.py)：
# -*- coding:utf-8 -*-
#*********************************************************#
writelines使用方法python# @@FileName: snakemqConnector.py
# @@Author:
Shelwinnee
# @@Create Date: 2016-04-14 22:24:14
# @@Modify Date: 2016-04-14 22:24:34
# @@Description:
# note:if message persistent is
necessary, please make sure you know the following instruction meaning, # 'sqlite object created in a thread can
only be used in that same thread'
# this problem's solution: modify the
connection method of sqlite3, open file ('snakemq\storage\sqlite.py') ,and
modify the following code:
# =
#*********************************************************#
import threading
import snakemq
import snakemq.link
import snakemq.packeter
ssaging
ssage
from snakemq.storage.sqlite import
SqliteQueuesStorage
ssage 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.link = None
self.snakemqident = snakemqident
self.pktr = None
self.persistent = persistent
<_recv = Callback()
self._initConnector()
def
run(self):
logger.info("connector
<")
if self.link != None:
self.link.loop()
logger.info("connector
<")
def
terminate(self):
logger.info("connetor
<")
if self.link != None:
self.link.stop()
self.link.cleanup()
logger.info("connetor
terminated")
def
on_recv_message(self, conn, ident, message):
try:
<_recv(ident,
message.data.decode('utf-8'))#dispatch received data
except Exception as e:
<("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
= ssage.Message(byteseq, ttl=60, flags=FLAG_PERSISTENT) else:
msg
= ssage.Message(byteseq, ttl=60)
ssaging == None:
<("connector:messaging
is not initialized, send message failed")
return
msg)
'''
'''
def
_initConnector(self):
try:
self.link
= snakemq.link.Link()
self.link.add_connector((Ip,
self.pktr
= snakemq.packeter.Packeter(self.link)
if
self.persistent:
storage =
SqliteQueuesStorage("SnakemqStorage.db")
else:
except Exception as e:
<("connector:{0}".format(e))
finally:
logger.info("connector[{0}]
".format(self.snakemqident))
Listener类(snakemqListener.py)：

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。