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

Python  /  管理员 发布于 7年前   230

一、关于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 nxloggerimport 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 threadingimport snakemqimport snakemq.linkimport snakemq.packeterimport snakemq.messagingimport snakemq.messagefrom snakemq.storage.sqlite import SqliteQueuesStoragefrom snakemq.message import FLAG_PERSISTENTfrom common.callbacks import Callback from common import nxloggerimport 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 threadingimport snakemqimport snakemq.linkimport snakemq.packeterimport snakemq.messagingimport snakemq.messagefrom common import nxloggerimport snakemqlogger as loggerfrom common.callbacks import Callbackclass 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 snakemqConnectorimport timeimport sysimport osdef 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 snakemqListenerimport 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进程间通信Queue消息队列用法分析
• 利用Python学习RabbitMQ消息队列
• Python中线程的MQ消息队列实现以及消息队列的优点解析
• 详解Python操作RabbitMQ服务器消息队列的远程结果返回
• Python操作RabbitMQ服务器实现消息队列的路由功能
• 利用Python操作消息队列RabbitMQ的方法教程
• Python RabbitMQ消息队列实现rpc
• python实现RabbitMQ的消息队列的示例代码
• Python多进程库multiprocessing中进程池Pool类的使用详解
• Python 多进程并发操作中进程池Pool的实例
• Python多进程池 multiprocessing Pool用法示例
• Python高级编程之消息队列(Queue)与进程池(Pool)实例详解