RabbitMQ からのデータを処理するために Pika を使用しています。さまざまな種類の問題に遭遇したように見えたので、切断をどのように処理できるかを確認するために、小さなテスト アプリケーションを作成することにしました。
次のことを行うこのテストアプリを作成しました。
- ブローカーに接続し、成功するまで再試行します
- 接続したら、キューを作成します。
- このキューを消費し、結果を python Queue.Queue(0) に入れます
- Queue.Queue(0) からアイテムを取得し、ブローカ キューに戻します。
私が気づいたのは2つの問題でした：
- 別のホスト (VM 内) の rabbitmq に接続している 1 つのホストからスクリプトを実行すると、このスクリプトはエラーを生成せずにランダムな瞬間に終了します。
- RabbitMQ がインストールされているのと同じホストでスクリプトを実行すると、スクリプトは正常に実行され、実行され続けます。
これは、ネットワークの問題が原因である可能性があります。接続はあまり堅牢ではありませんが、パケットがドロップされました。
スクリプトが RabbitMQ サーバーでローカルに実行され、RabbitMQ を強制終了すると、スクリプトは次のエラーで終了します: "ERROR pika SelectConnection: Socket Error on 3: 104"
そのため、再接続戦略を正常に機能させることができないようです。誰かがコードを見て、私が間違っていることを確認できますか?
ありがとう、
ジェイ
#!/bin/python
import logging
import threading
import Queue
import pika
from pika.reconnection_strategies import SimpleReconnectionStrategy
from pika.adapters import SelectConnection
import time
from threading import Lock
class Broker(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.logging = logging.getLogger(__name__)
self.to_broker = Queue.Queue(0)
self.from_broker = Queue.Queue(0)
self.parameters = pika.ConnectionParameters(host='sandbox',heartbeat=True)
self.srs = SimpleReconnectionStrategy()
self.properties = pika.BasicProperties(delivery_mode=2)
self.connection = None
while True:
try:
self.connection = SelectConnection(self.parameters, self.on_connected, reconnection_strategy=self.srs)
break
except Exception as err:
self.logging.warning('Cant connect. Reason: %s' % err)
time.sleep(1)
self.daemon=True
def run(self):
while True:
self.submitData(self.from_broker.get(block=True))
pass
def on_connected(self,connection):
connection.channel(self.on_channel_open)
def on_channel_open(self,new_channel):
self.channel = new_channel
self.channel.queue_declare(queue='sandbox', durable=True)
self.channel.basic_consume(self.processData, queue='sandbox')
def processData(self, ch, method, properties, body):
self.logging.info('Received data from broker')
self.channel.basic_ack(delivery_tag=method.delivery_tag)
self.from_broker.put(body)
def submitData(self,data):
self.logging.info('Submitting data to broker.')
self.channel.basic_publish(exchange='',
routing_key='sandbox',
body=data,
properties=self.properties)
if __name__ == '__main__':
format=('%(asctime)s %(levelname)s %(name)s %(message)s')
logging.basicConfig(level=logging.DEBUG, format=format)
broker=Broker()
broker.start()
try:
broker.connection.ioloop.start()
except Exception as err:
print err