快捷導(dǎo)航

Python的消息隊(duì)列包SnakeMQ使用初探

更新時(shí)間：2016年06月29日 17:35:01 投稿：goldensun

使用消息隊(duì)列在數(shù)據(jù)的通信中擁有很多優(yōu)點(diǎn),SnakeMQ是一個(gè)開(kāi)源的用Python實(shí)現(xiàn)的跨平臺(tái)MQ庫(kù),well,Python的消息隊(duì)列包SnakeMQ使用初探,here we go:

一、關(guān)于snakemq的官方介紹
SnakeMQ的GitHub項(xiàng)目頁(yè)：https://github.com/dsiroky/snakemq
1.純python實(shí)現(xiàn)，跨平臺(tái)

2.自動(dòng)重連接

3.可靠發(fā)送--可配置的消息方式與消息超時(shí)方式

4.持久化/臨時(shí) 兩種隊(duì)列

5.支持異步 -- poll()

6.symmetrical -- 單個(gè)TCP連接可用于雙工通訊

7.多數(shù)據(jù)庫(kù)支持 -- SQLite、MongoDB……

8.brokerless - 類(lèi)似ZeroMQ的實(shí)現(xiàn)原理

9.擴(kuò)展模塊：RPC, bandwidth throttling

以上都是官話，需要自己驗(yàn)證，動(dòng)手封裝了一下，感覺(jué)萌萌噠。

二、幾個(gè)主要問(wèn)題說(shuō)明

1.支持自動(dòng)重連，不需要自己動(dòng)手寫(xiě)心跳邏輯，你只需要關(guān)注發(fā)送和接收就行

2.支持?jǐn)?shù)據(jù)持久化，如果開(kāi)始持久化，在重連之后會(huì)自動(dòng)發(fā)送數(shù)據(jù)。

3.數(shù)據(jù)的接收，snakemq通過(guò)提供回調(diào)實(shí)現(xiàn)，你只需要寫(xiě)個(gè)接收方法添加到回調(diào)列表里去。

4.數(shù)據(jù)的發(fā)送，在此發(fā)送的都是bytes類(lèi)型（二進(jìn)制），因此需要轉(zhuǎn)換。我在程序中測(cè)試的都是文本字符串，使用str.encode(‘utf-8')轉(zhuǎn)換成bytes，接收時(shí)再轉(zhuǎn)換回來(lái)。

5.術(shù)語(yǔ)解釋?zhuān)珻onnector：類(lèi)似于socket的TcpClient，Lisenter：類(lèi)似于socket的TcpServer，每個(gè)connector或者listener都一個(gè)一個(gè)ident標(biāo)識(shí)，發(fā)送和接收數(shù)據(jù)時(shí)就知道是誰(shuí)的數(shù)據(jù)了。

6.使用sqlite持久化時(shí)，需要修改源碼，sqlite3.connect(filename,check_same_thread = False)，用于解決多線程訪問(wèn)sqlite的問(wèn)題。（會(huì)不會(huì)死鎖？）

7.啟動(dòng)持久化時(shí)，如果重新連上，則會(huì)自動(dòng)發(fā)送，保證可靠。

8.為了封裝的需要，數(shù)據(jù)接收以后，我通過(guò)callback方式傳送出去。

三、代碼

說(shuō)明代碼中使用了自定義的日志模塊

from common import nxlogger

import snakemqlogger as logger

可替換成logging的。

回調(diào)類(lèi)（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類(lèi)（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類(lèi)（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)

測(cè)試代碼connector（testSnakeConnector.py）：

讀取本地一個(gè)1M的文件，然后發(fā)送給listener，然后listener發(fā)回一個(gè)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()   

 

測(cè)試代碼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...")

您可能感興趣的文章: