rabbitmq

rabbitmq安装:

1.安装rabbitmq-server

　　yum -y install rabbitmq-server
2.启动rabbitmq-server

　　systemctl start rabbitmq-server 

3.查看rabbitmq-server队列

rabbitmqctl list_queues 
4.添加用户名密码
　　rabbitmqctl add_user alex alex3714 
5.给用户授权
　　rabbitmqctl set_permissions -p / alex ".*" ".*" ".*" 

一、实现最简单的队列通信

发送端:

import pika#导入连接rabbitmq的模块

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.117.130'))#创建一个阻塞连接对象,ConnectionParameters连接参数+ip

channel = connection.channel()#连接通道,建立的rabbit协议的通信

#声明queue
channel.queue_declare(queue='hello')#声明队列的名字hello
#RabbitMQ消息不能直接发送到队列，它总是需要通过一个交换。
#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()

接收端

import pika

credential = pika.PlainCredentials('alex','alex3714')#用户认证
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.117.130'))

channel = connection.channel()#连接通道,建立的rabbit协议的通信
"""
你可能会问为什么我们宣布再排队‒我们已经在我们以前的代码声明。如果我们确信队列已经存在，我们可以避免这种情况。例如，如果send.py程序
运行之前。但我们还不能确定首先运行哪个程序。在这种情况下，这是一件好事。在两个程序中重复声明队列的实践。
"""
#通道的实例
channel.queue_declare(queue='hello')

#定义一个回调函数
def callback(ch,method,properties,body):
    print("[x] Received %r" %body)

#收到消息就调用这个
channel.basic_consume(callback,
                      queue='hello',
                      no_ack=True)
print('[*] Waiting for messages. To exit press Ctrl+C')
channel.start_consuming()#开始消息,是个死循环,一直监听消息

 在linux系统中，通过: rabbitmqctl list_queues  查看消息。

二、Work Queues （一个发消息，两个收消息，收消息是公平的依次分发）

在这种模式下，RabbitMQ会默认把P发的消息依次分发给各个消费者(C1和C2),跟负载均衡差不多。

消息提供者代码

import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.117.130'))#服务端创建阻塞连接的对象(pika的连接参数ip)
channel = connection.channel()

#声明queue
channel.queue_declare(queue='task_queue')#声明一个队列
#RabbitMQ消息不能直接发送到队列，它总是需要通过一个交换。
import sys
# m = sys.argv
# print(m)#['F:/project/rabbitmq_test/工作队列work queue/消费提供者.py']
message = ' '.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()

消费者代码

import pika,time
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.117.130'))#客户端创建阻塞连接的对象(pika的连接参数ip)
channel = connection.channel()#通道，对象.方法执行对应的函数

def callback(ch,method,properties,body):#method:请求方式,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()#监听阻塞监听队列

此时，先启动消息生产者，然后再分别启动3个消费者，通过生产者多发送几条消息，你会发现，这几条消息会被依次分配到各个消费者身上

运行结果:

启动消费提供者:

启动第一个消费者:

启动第二个消费者

启动第三个消费者

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.

执行任务可能需要几秒钟。你可能会想，如果一个消费者开始一个长期的任务，并且仅仅部分地完成它，就会发生什么。使用我们当前的代码，一旦RabbitMQ向客户发送消息，它将立即将其从内存中删除。在这种情况下，如果你杀死一个工作人员，我们将丢失正在处理的消息。
我们还会丢失所有发送给该特定工作人员但尚未处理的消息。

但是我们不想失去任何任务。如果一个工人死亡，我们希望把这个任务交给另一个工作人员。

为了确保消息永远不会丢失，RabbitMQ支持消息确认。从消费者那里发送一个确认信息（告示），告诉RabbitMQ已经收到，处理了一个特定的消息，并且RabbitMQ可以自由删除它。

如果消费者死机（其通道关闭，连接关闭或TCP连接丢失），而不发送确认信息，RabbitMQ将会明白消息未被完全处理并重新排队。如果同时有其他消费者在线，则会迅速将其重新提供给另一个消费者。这样就可以确保没有消息丢失，即使工作人员偶尔也会死亡。

没有任何消息超时;当消费者死亡时，RabbitMQ将重新发送消息。即使处理消息需要非常长的时间，这很好。

消息确认默认情况下打开。在以前的例子中，我们通过no_ack = True标志来明确地将它们关闭。一旦完成任务，现在该删除这个标志并发送正确的确认。

回调函数:

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。

 

使用这个代码，我们可以确定即使在处理消息时，使用CTRL + C杀死一个消费者，也不会丢失任何东西。 消费者死亡之后不久，所有未确认的消息将被重新发送。

三、消息持久化

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:

我们已经学会如何确保即使消费者死亡，任务也不会丢失（默认情况下，如果要禁用使用no_ack = True）。 但是如果RabbitMQ服务器停止，我们的任务仍然会丢失。

当RabbitMQ退出或崩溃时，它会忘记队列和消息，除非你不告诉它。 需要两件事来确保消息不会丢失：我们需要将队列和消息标记为耐用。

首先，我们需要确保RabbitMQ不会失去我们的队列。 为了这样做，我们需要将其声明为持久的：

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:

虽然这个命令本身是正确的，但它在我们的设置中将不起作用。 这是因为我们已经定义了一个不耐用的名为hello的队列。 RabbitMQ不允许您重新定义具有不同参数的现有队列，并会向尝试执行此操作的任何程序返回错误。

 但是有一个快速的解决方法 - 让我们声明一个具有不同名称的队列，例如：ttask_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.

这个queue_declare更改需要应用于生产者和消费者代码。在这一点上，我们确信，即使RabbitMQ重新启动，task_queue队列也不会丢失。 现在我们需要通过提供一个值为2的delivery_mode属性将我们的消息标记为持久性。

channel.basic_publish(exchange='',

                      routing_key="task_queue",

                      body=message,

                      properties=pika.BasicProperties(

                         delivery_mode = 2, # make message persistent  #使消息持久化

                      ))

四、消息公平分发

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

channel.basic_qos(prefetch_count=1)

生产者端带消息持久化+公平分发的完整代码