一、關(guān)于snakemq的官方介紹
SnakeMQ的GitHub項(xiàng)目頁：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ù)庫支持 -- SQLite、MongoDB……
8.brokerless - 類似ZeroMQ的實(shí)現(xiàn)原理
9.擴(kuò)展模塊：RPC, bandwidth throttling
以上都是官話，需要自己驗(yàn)證，動(dòng)手封裝了一下，感覺萌萌噠。
二、幾個(gè)主要問題說明
1.支持自動(dòng)重連，不需要自己動(dòng)手寫心跳邏輯，你只需要關(guān)注發(fā)送和接收就行
2.支持?jǐn)?shù)據(jù)持久化，如果開始持久化，在重連之后會(huì)自動(dòng)發(fā)送數(shù)據(jù)。
3.數(shù)據(jù)的接收，snakemq通過提供回調(diào)實(shí)現(xiàn)，你只需要寫個(gè)接收方法添加到回調(diào)列表里去。
4.數(shù)據(jù)的發(fā)送，在此發(fā)送的都是bytes類型（二進(jìn)制），因此需要轉(zhuǎn)換。我在程序中測(cè)試的都是文本字符串，使用str.encode(‘utf-8')轉(zhuǎn)換成bytes，接收時(shí)再轉(zhuǎn)換回來。
5.術(shù)語解釋，Connector：類似于socket的TcpClient，Lisenter：類似于socket的TcpServer，每個(gè)connector或者listener都一個(gè)一個(gè)ident標(biāo)識(shí)，發(fā)送和接收數(shù)據(jù)時(shí)就知道是誰的數(shù)據(jù)了。
6.使用sqlite持久化時(shí)，需要修改源碼，sqlite3.connect(filename,check_same_thread = False)，用于解決多線程訪問sqlite的問題。（會(huì)不會(huì)死鎖？）
7.啟動(dòng)持久化時(shí)，如果重新連上，則會(huì)自動(dòng)發(fā)送，保證可靠。
8.為了封裝的需要，數(shù)據(jù)接收以后，我通過callback方式傳送出去。
三、代碼
說明代碼中使用了自定義的日志模塊
from common import nxlogger
import snakemqlogger as logger
可替換成logging的。
回調(diào)類（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)
測(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...")
更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號(hào)聯(lián)系： 360901061
您的支持是博主寫作最大的動(dòng)力，如果您喜歡我的文章，感覺我的文章對(duì)您有幫助，請(qǐng)用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點(diǎn)擊下面給點(diǎn)支持吧，站長非常感激您！手機(jī)微信長按不能支付解決辦法：請(qǐng)將微信支付二維碼保存到相冊(cè)，切換到微信，然后點(diǎn)擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對(duì)您有幫助就好】元
