您的位置:首页 > 编程语言 > Python开发

Python的消息队列包SnakeMQ使用初探

2016-06-29 17:35 871 查看

一、关于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 Callbackfrom common import nxlogger
import snakemqlogger as loggerclass 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.messagefrom common import nxlogger
import snakemqlogger as loggerfrom 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...")

您可能感兴趣的文章:

内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签:  Python 消息队列