11

送信しようとしているメッセージを消費するコンシューマー/ワーカーが存在するかどうかを確認したいと思います。

Workerがない場合は、いくつかのワーカーを開始し (コンシューマーとパブリッシャーの両方が 1 台のマシン上にあります)、次にMessagesのパブリッシュに取り掛かります。

のような関数がある場合connection.check_if_has_consumers、私はそれを次のように実装します-

import pika
import workers

# code for publishing to worker queue
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# if there are no consumers running (would be nice to have such a function)
if not connection.check_if_has_consumers(queue="worker_queue", exchange=""):
    # start the workers in other processes, using python's `multiprocessing`
    workers.start_workers()

# now, publish with no fear of your queues getting filled up
channel.queue_declare(queue="worker_queue", auto_delete=False, durable=True)
channel.basic_publish(exchange="", routing_key="worker_queue", body="rockin",
                            properties=pika.BasicProperties(delivery_mode=2))
connection.close()

check_if_has_consumersしかし、pikaで機能する関数が見つかりません。

pikaを使用してこれを達成する方法はありますか? それとも、うさぎと直接話すことで？

完全にはわかりませんが、RabbitMQはさまざまなキューにサブスクライブしているコンシューマーの数を認識していると思います。メッセージをそれらにディスパッチし、 ACKを受け入れるからです。

3時間前にRabbitMQを始めたばかりです...どんな助けも大歓迎です...

これが私が書いたworkers.pyコードです。

import multiprocessing
import pika


def start_workers(num=3):
    """start workers as non-daemon processes"""
    for i in xrange(num):    
        process = WorkerProcess()
        process.start()


class WorkerProcess(multiprocessing.Process):
    """
    worker process that waits infinitly for task msgs and calls
    the `callback` whenever it gets a msg
    """
    def __init__(self):
        multiprocessing.Process.__init__(self)
        self.stop_working = multiprocessing.Event()

    def run(self):
        """
        worker method, open a channel through a pika connection and
        start consuming
        """
        connection = pika.BlockingConnection(
                              pika.ConnectionParameters(host='localhost')
                     )
        channel = connection.channel()
        channel.queue_declare(queue='worker_queue', auto_delete=False,
                                                    durable=True)

        # don't give work to one worker guy until he's finished
        channel.basic_qos(prefetch_count=1)
        channel.basic_consume(callback, queue='worker_queue')

        # do what `channel.start_consuming()` does but with stopping signal
        while len(channel._consumers) and not self.stop_working.is_set():
            channel.transport.connection.process_data_events()

        channel.stop_consuming()
        connection.close()
        return 0

    def signal_exit(self):
        """exit when finished with current loop"""
        self.stop_working.set()

    def exit(self):
        """exit worker, blocks until worker is finished and dead"""
        self.signal_exit()
        while self.is_alive(): # checking `is_alive()` on zombies kills them
            time.sleep(1)

    def kill(self):
        """kill now! should not use this, might create problems"""
        self.terminate()
        self.join()


def callback(channel, method, properties, body):
    """pika basic consume callback"""
    print 'GOT:', body
    # do some heavy lifting here
    result = save_to_database(body)
    print 'DONE:', result
    channel.basic_ack(delivery_tag=method.delivery_tag)

編集：

先に進まなければならないので、より良いアプローチがない限り、私が取ろうとしている回避策は次のとおりです。

したがって、RabbitMQにはこれらのHTTP 管理 APIがあり、管理プラグインを有効にした後に機能し、HTTP API ページの中央に

/api/connections - 開いているすべての接続のリスト。

/api/connections/name - 個々の接続。それを削除すると、接続が閉じます。

したがって、ワーカーとプロデュースの両方を異なる接続名/ユーザーで接続すると、ワーカー接続が開いているかどうかを確認できます... (ワーカーが停止すると問題が発生する可能性があります...)

より良い解決策を待っています...

編集：

これはrabbitmqのドキュメントで見つけたばかりですが、これをPythonで行うのはハックです:

shobhit@oracle:~$ sudo rabbitmqctl -p vhostname list_queues name consumers
Listing queues ...
worker_queue    0
...done.

だから私は次のようなことができます

subprocess.call("echo password|sudo -S rabbitmqctl -p vhostname list_queues name consumers | grep 'worker_queue'")

ハッキー...まだピカがこれを行うためのPython関数を持っていることを願っています...

ありがとう、

4

2 に答える 2

7

私もこれを調べていました。ソースとドキュメントを読んだ後、channel.pyで次のことに出くわしました。

@property
def consumer_tags(self):
    """Property method that returns a list of currently active consumers

    :rtype: list

    """
    return self._consumers.keys()

私自身のテストは成功しました。チャネルオブジェクトがself._channelである場合、次を使用しました。

if len(self._channel.consumer_tags) == 0:
        LOGGER.info("Nobody is listening.  I'll come back in a couple of minutes.")
        ...
于 2012-12-12T20:33:40.800 に答える
0

私は実際に別の問題を探して偶然これを見つけましたが、Basic_Publish 関数には、デフォルトで False に設定されているパラメーター「Immediate」があります。

実行できる 1 つのアイデアは、Immediate フラグを True に設定することです。これにより、キューに置かれるのではなく、コンシューマーによってすぐに消費される必要があります。ワーカーがメッセージを消費できない場合、別のワーカーを開始するように指示するエラーが返されます。

システムのスループットに応じて、これは多くの余分なワーカーを生成するか、死んだワーカーを置き換えるためにワーカーを生成します。前者の問題については、制御キューを介してワーカーを単純に追跡する管理者のようなシステムを作成できます。このシステムでは、「ランナー」のようなプロセスに、不要になったワーカーのプロセスを強制終了するように指示できます。

于 2012-11-12T14:59:26.200 に答える