点击下载
本文文档

当前位置：首页 - 科技 - 知识百科 - 正文

Python的消息队列包SnakeMQ使用

来源：动视网责编：小采时间：2020-11-27 14:26:18

Python的消息队列包SnakeMQ使用

Python的消息队列包SnakeMQ使用:使用消息队列在数据的通信中拥有很多优点,SnakeMQ是一个开源的用Python实现的跨平台MQ库,well,Python的消息队列包SnakeMQ使用初探,here we go:一、关于snakemq的官方介绍SnakeMQ的GitHub项目页：https://github.com/dsiroky/sn

推荐度：

点击下载本文 文档为doc格式

导读Python的消息队列包SnakeMQ使用:使用消息队列在数据的通信中拥有很多优点,SnakeMQ是一个开源的用Python实现的跨平台MQ库,well,Python的消息队列包SnakeMQ使用初探,here we go:一、关于snakemq的官方介绍SnakeMQ的GitHub项目页：https://github.com/dsiroky/sn

使用消息队列在数据的通信中拥有很多优点,SnakeMQ是一个开源的用Python实现的跨平台MQ库,well,Python的消息队列包SnakeMQ使用初探,here we go:

一、关于snakemq的官方介绍
SnakeMQ的GitHub项目页：https://github.com/dsiroky/snakemq1.纯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...")

更多Python的消息队列包SnakeMQ使用相关文章请关注PHP中文网！

Python的消息队列包SnakeMQ使用

推荐度：

点击下载本文 文档为doc格式

标签：的消息 python Pythonic

热门焦点

Python的消息队列包SnakeMQ使用

Python的消息队列包SnakeMQ使用

Python的消息队列包SnakeMQ使用

最新推荐

猜你喜欢

热门推荐