メッセージ キューの使用には、データ通信において多くの利点があります。SnakeMQ は、Python で実装されたオープン ソースのクロスプラットフォーム MQ ライブラリです。それでは、Python のメッセージ キュー パッケージ SnakeMQ の使用に関する予備調査を行ってみましょう。
1. 概要Snakemq 公式紹介
SnakeMQ の GitHub プロジェクト ページ: https://github.com/dsiroky/snakemq1. 純粋な Python 実装、クロスプラットフォーム
2. 信頼性の高い送信 - 構成可能なメッセージ メソッドとメッセージ タイムアウト メソッド
4. 永続的/一時的な 2 つのキュー
5. 非同期のサポート -- poll()
6. 単一の TCP 接続を二重通信に使用できます
7. 複数のデータベースのサポート -- SQLite、MongoDB。 ..
8.brokerless - ZeroMQの実装原理に似ています
9.拡張モジュール: RPC、帯域幅調整
上記はすべて公式の言葉であり、自分でパッケージ化して感じました。かわいい。
1. 自動再接続をサポートし、ハートビート ロジックを自分で記述する必要がなく、送信と受信のみに集中する必要があります
2.持続性があり、再接続後にデータが自動的に送信されます。
3. Snakemq はコールバックを提供することでデータ受信を実装します。必要なのは、受信メソッドを作成してコールバック リストに追加することだけです。
4. データはバイト型（バイナリ）で送信されるため、変換する必要があります。プログラムでテストするのはすべてテキスト文字列です。str.encode('utf-8') を使用してバイトに変換し、受信時に元に戻します。
5. 用語の説明、コネクタ: ソケットに似た TcpClient、リスナー: ソケットに似た TcpServer それぞれのコネクタまたはリスナーは、データを送受信するときに、それが誰のデータであるかを知ることができます。
6. sqlite 永続性を使用する場合、sqlite へのマルチスレッド アクセスの問題を解決するには、ソース コード sqlite3.connect(filename, check_same_thread = False) を変更する必要があります。 (デッドロックは発生しますか?)
7. 永続化を開始するとき、接続が再接続された場合は、信頼性を確保するために自動的に送信されます。
8. カプセル化の目的で、データを受信した後、コールバックを通じて送信します。
説明 このコードは、logging に置き換えることができるカスタム ログ モジュール
from common import nxlogger import snakemqlogger as logger
を使用します。
コールバック クラス (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)コネクタ クラス (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))リスナー クラス (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)テストコードコネクタ(testSnakeConnector.py):
ローカルの 1M ファイルを読み取り、それをリスナーに送信すると、リスナーは 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 中国語 Web サイトに注目してください。
![[Web フロントエンド] Node.js クイック スタート](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
