我想检查是否存在 Consumer/Worker 来消费我将要发送的 Message。
如果没有任何 Worker,我会启动一些 worker(消费者和发布者都在一台机器上)然后开始发布消息。 p>
如果有像 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()
但我无法在 pika 中找到具有 check_if_has_consumers
功能的任何函数。
有什么方法可以使用 pika 来完成这个任务吗?或者,通过直接与兔子交谈?
我不完全确定,但我真的认为 RabbitMQ 会知道订阅不同队列的消费者数量,因为它会向他们发送消息并接受确认
我 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 management apis ,它们在您打开 management plugin 后工作在 HTTP api 页面的中间有
/api/connections - A list of all open connections.
/api/connections/name - An individual connection. DELETEing it will close the connection.
因此,如果我通过不同的Connection 名称/用户连接我的Workers 和我的Produces,我将能够检查是否Worker Connection 已打开...(当 worker 死亡时可能会出现问题...)
将等待更好的解决方案...
编辑:
刚刚在 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'")
hacky...仍然希望 pika 有一些 python 函数来做到这一点...
谢谢,
最佳答案
我也在研究这个。阅读源代码和文档后,我在 channel.py 中发现了以下内容:
@property
def consumer_tags(self):
"""Property method that returns a list of currently active consumers
:rtype: list
"""
return self._consumers.keys()
我自己测试成功了。我在 channel 对象为 self._channel 的地方使用了以下内容:
if len(self._channel.consumer_tags) == 0:
LOGGER.info("Nobody is listening. I'll come back in a couple of minutes.")
...
关于python - 在 Pika 或 RabbitMQ 中,如何检查当前是否有消费者正在消费?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/13037121/