rabbitmq使用方法(二)
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了rabbitmq使用方法(二),小编现在分享给大家,供广大互联网技能从业者学习和参考。文章包含5943字,纯文字阅读大概需要9分钟。
内容图文
![rabbitmq使用方法(二)](/upload/InfoBanner/zyjiaocheng/1333/e82a6b4f2ceb454a90954fdb87cd270f.jpg)
Work Queues
![技术分享](/upload/getfiles/default/2022/11/14/20221114060142340.jpg)
In the first tutorial we wrote programs to send and receive messages from a named queue. In this one we‘ll create a Work Queue that will be used to distribute time-consuming tasks among multiple workers.
The main idea behind Work Queues (aka: Task Queues) is to avoid doing a resource-intensive task immediately and having to wait for it to complete. Instead we schedule the task to be done later. We encapsulate a task as a message and send it to the queue. A worker process running in the background will pop the tasks and eventually execute the job. When you run many workers the tasks will be shared between them.
Round-robin dispatching
One of the advantages of using a Task Queue is the ability to easily parallelise work. If we are building up a backlog of work, we can just add more workers and that way, scale easily.
By default, RabbitMQ will send each message to the next consumer, in sequence. On average every consumer will get the same number of messages. This way of distributing messages is called round-robin.
Message acknowledgment
Doing a task can take a few seconds. You may wonder what happens if one of the consumers starts a long task and dies with it only partly done. With our current code once RabbitMQ delivers message to the customer it immediately removes it from memory. In this case, if you kill a worker we will lose the message it was just processing. We‘ll also lose all the messages that were dispatched to this particular worker but were not yet handled.
But we don‘t want to lose any tasks. If a worker dies, we‘d like the task to be delivered to another worker.
In order to make sure a message is never lost, RabbitMQ supports message acknowledgments. An ack(nowledgement) is sent back from the consumer to tell RabbitMQ that a particular message had been received, processed and that RabbitMQ is free to delete it.
If consumer dies without sending an ack, RabbitMQ will understand that a message wasn‘t processed fully and will redeliver it to another consumer. That way you can be sure that no message is lost, even if the workers occasionally die.
There aren‘t any message timeouts; RabbitMQ will redeliver the message only when the worker connection dies. It‘s fine even if processing a message takes a very, very long time.
Message durability
We have learned how to make sure that even if the consumer dies, the task isn‘t lost. But our tasks will still be lost if RabbitMQ server stops.
When RabbitMQ quits or crashes it will forget the queues and messages unless you tell it not to. Two things are required to make sure that messages aren‘t lost: we need to mark both the queue and messages as durable.
First, we need to make sure that RabbitMQ will never lose our queue. In order to do so, we need to declare it as durable:
1 channel.queue_declare(queue=‘hello‘, durable=True)
Although this command is correct by itself, it won‘t work in our setup. That‘s because we‘ve already defined a queue called hello which is not durable. RabbitMQ doesn‘t allow you to redefine an existing queue with different parameters and will return an error to any program that tries to do that. But there is a quick workaround - let‘s declare a queue with different name
This queue_declare change needs to be applied to both the producer and consumer code.
At that point we‘re sure that the task_queue queue won‘t be lost even if RabbitMQ restarts. Now we need to mark our messages as persistent - by supplying a delivery_mode property with a value 2.
1 channel.basic_publish(exchange=‘‘, 2 routing_key="task_queue", 3 body=message, 4 properties=pika.BasicProperties( 5 delivery_mode = 2, # make message persistent6 ))
Fair dispatch
You might have noticed that the dispatching still doesn‘t work exactly as we want. For example in a situation with two workers, when all odd messages are heavy and even messages are light, one worker will be constantly busy and the other one will do hardly any work. Well, RabbitMQ doesn‘t know anything about that and will still dispatch messages evenly.
This happens because RabbitMQ just dispatches a message when the message enters the queue. It doesn‘t look at the number of unacknowledged messages for a consumer. It just blindly dispatches every n-th message to the n-th consumer.
![技术分享](/upload/getfiles/default/2022/11/14/20221114060143710.jpg)
In order to defeat that we can use the basic.qos method with the prefetch_count=1 setting. This tells RabbitMQ not to give more than one message to a worker at a time. Or, in other words, don‘t dispatch a new message to a worker until it has processed and acknowledged the previous one. Instead, it will dispatch it to the next worker that is not still busy.
1 channel.basic_qos(prefetch_count=1)
Putting it all together
Final code of our new_task.py script:
1 # !/usr/bin/env python 2 import pika 3 import sys 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host=‘localhost‘)) 7 channel = connection.channel() 8 9 channel.queue_declare(queue=‘task_queue‘, durable=True) 1011 message = ‘‘.join(sys.argv[1:]) or"Hello World!"12 channel.basic_publish(exchange=‘‘, 13 routing_key=‘task_queue‘, 14 body=message, 15 properties=pika.BasicProperties( 16 delivery_mode = 2, # make message persistent17 )) 18print" [x] Sent %r" % (message,) 19 connection.close()
And our worker: (new_task.py source)
1 # !/usr/bin/env python 2 import pika 3 import time 4 5 connection = pika.BlockingConnection(pika.ConnectionParameters( 6 host=‘localhost‘)) 7 channel = connection.channel() 8 9 channel.queue_declare(queue=‘task_queue‘, durable=True) 10print‘ [*] Waiting for messages. To exit press CTRL+C‘1112def callback(ch, method, properties, body): 13print" [x] Received %r" % (body,) 14 time.sleep( body.count(‘.‘) ) 15print" [x] Done"16 ch.basic_ack(delivery_tag = method.delivery_tag) 1718 channel.basic_qos(prefetch_count=1) 19channel.basic_consume(callback, 20 queue=‘task_queue‘) 2122 channel.start_consuming()
原文:http://www.cnblogs.com/lintong/p/4382412.html
内容总结
以上是互联网集市为您收集整理的rabbitmq使用方法(二)全部内容,希望文章能够帮你解决rabbitmq使用方法(二)所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错,欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 gblab@vip.qq.com 举报,一经查实,本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。