Rabbitmq编程

 

 

实现最简单的队列通信

 

send端

#!/usr/bin/env pythonimport pikacredentials = pika.PlainCredentials("用户名","密码")connection = pika.BlockingConnection(pika.ConnectionParameters(    'localhost',credentials=credentials))channel = connection.channel() #建立了rabbit协议的通道# 声明queuechannel.queue_declare(queue='hello')# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.channel.basic_publish(exchange='',                      routing_key='hello',                      body='Hello World!')print(" [x] Sent 'Hello World!'")connection.close()

 

receive端

# _*_coding:utf-8_*___author__ = 'Alex Li'import pikacredentials = pika.PlainCredentials("用户名","密码")connection = pika.BlockingConnection(pika.ConnectionParameters(    'localhost',credentials=credentials))channel = connection.channel() #建立了rabbit协议的通道# You may ask why we declare the queue again ‒ we have already declared it in our previous code.# We could avoid that if we were sure that the queue already exists. For example if send.py program# was run before. But we're not yet sure which program to run first. In such cases it's a good# practice to repeat declaring the queue in both programs.channel.queue_declare(queue='hello')def callback(ch, method, properties, body):    print(" [x] Received %r" % body)# callback函数当拿到队列里的值，则调用channel.basic_consume(callback,                      queue='hello',                      no_ack=True)print(' [*] Waiting for messages. To exit press CTRL+C')channel.start_consuming()

 

#注意：远程连接rabbitmq server的话，需要配置权限。 #1.设置用户与密码

# > rabbitmqctl add_user name pass# > rabbitmqctl set_user_tags name administrator

#2.设置权限，允许从外面访问

# rabbitmqctl set_permissions -p /name ".*" ".*" ".*"

set_permissions [-p vhost] {user} {conf} {write} {read}vhostThe name of the virtual host to which to grant the user access, defaulting to /.userThe name of the user to grant access to the specified virtual host.confA regular expression matching resource names for which the user is granted configure permissions.writeA regular expression matching resource names for which the user is granted write permissions.readA regular expression matching resource names for which the user is granted read permissions.

#3.生产者与消费者添加认证信息

credentials = pika.PlainCredentials("用户名","密码")

 

#为什么要声明两次queue，这里hello为队列名# channel.queue_declare(queue='hello')# 解决发起者先启动，而接收者还没有启动，发送者先创建queue，# 如果发起者已经声明了，接收者会检测有没有queue，如果有了，实际接收者是不会执行声明的，没有就会声明这个queue。

　　

 

消息公平分发

在这种模式下，RabbitMQ会默认把p发的消息依次分发给各个消费者(c)。

轮巡公平的发送给接收者，比如第一次发送给第一个接收者，第二次发送给第二格接受者，如此。

send端

import pikaimport timecredentials = pika.PlainCredentials("用户名","密码")connection = pika.BlockingConnection(pika.ConnectionParameters(    'localhost',credentials=credentials))channel = connection.channel()# 声明queuechannel.queue_declare(queue='task_queue')# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.import sysmessage = ' '.join(sys.argv[1:]) or "Hello World! %s" % time.time()channel.basic_publish(exchange='',                      routing_key='task_queue',                      body=message,                      properties=pika.BasicProperties(                          delivery_mode=2,  # make message persistent                      )                      )print(" [x] Sent %r" % message)connection.close()

 

receive端

# _*_coding:utf-8_*_import pika, timecredentials = pika.PlainCredentials("用户名","密码")connection = pika.BlockingConnection(pika.ConnectionParameters(    'localhost',credentials=credentials))channel = connection.channel()def callback(ch, method, properties, body):    print(" [x] Received %r" % body)    time.sleep(20)    print(" [x] Done")    print("method.delivery_tag", method.delivery_tag)    ch.basic_ack(delivery_tag=method.delivery_tag)channel.basic_consume(callback,                      queue='task_queue',                      no_ack=True                      )print(' [*] Waiting for messages. To exit press CTRL+C')channel.start_consuming()

 

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 a consumer dies (its channel is closed, connection is closed, or TCP connection is lost) without sending an ack, RabbitMQ will understand that a message wasn't processed fully and will re-queue it. If there are other consumers online at the same time, it will then quickly 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 when the consumer dies. It's fine even if processing a message takes a very, very long time.

Message acknowledgments are turned on by default. In previous examples we explicitly turned them off via the no_ack=True flag. It's time to remove this flag and send a proper acknowledgment from the worker, once we're done with a task.

def callback(ch, method, properties, body):    print " [x] Received %r" % (body,)    time.sleep( body.count('.') )    print " [x] Done"    ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_consume(callback,                      queue='hello')

　　Using this code we can be sure that even if you kill a worker using CTRL+C while it was processing a message, nothing will be lost. Soon after the worker dies all unacknowledged messages will be redelivered

 

 

消息持久化　　

We have learned how to make sure that even if the consumer dies, the task isn't lost(by default, if wanna disable  use no_ack=True). 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:

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, for exampletask_queue:

channel.queue_declare(queue='task_queue', durable=True)

 

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.

channel.basic_publish(exchange='',                      routing_key="task_queue",                      body=message,                      properties=pika.BasicProperties(                         delivery_mode = 2, # make message persistent                      ))

 

 

负载均衡

如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，可以在各个消费者端， 配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了。

send端

receive端