rabbitMQ玩耍日记

2019-10-25 本文已影响0人魂狩

rabbitMQ是个啥东东，想要玩的人自然已经有了初步了解。这里直接开始安装。安装环境为ubuntu 18.04。

ubuntu自带rabbitMQ的包，直接apt安装即可。apt install rabbitmq-server。安装时会把rabbitmq和其依赖erlang一起安装。
默认情况下web管理界面是不会开启的，需要手动启动。rabbitmq-plugins enable rabbitmq_management。
启动后默认端口为15672，默认用户guest/guest。登录进去后可以先创建自己的帐号，然后将默认帐号删除;也可以直接使用命令行进行操作：
1. rabbitmqctl add_user small 123456 创建一个密码是123456的small用户
2. rabbitmqctl set_user_tags small administrator 给此用户授管理员权限。
创建用户后需要给用户权限。根据情况给用户分配虚拟主机（virtual hosts），用户只能在对应虚拟主机进行操作。
分配权限后就可以愉快玩耍了。以下为最简单的demo，来自官方文档：

#生产者
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(credentials=pika.PlainCredentials('small','123456')))
channel = connection.channel()
channel.queue_declare(queue='test')
channel.basic_publish(exchange='', routing_key='test', body='hello world')
connection.close()

#消费者
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(credentials=pika.PlainCredentials('small','123456')))
channel = connection.channel()
channel.queue_declare(queue='test')
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

channel.basic_consume(
    queue='test', on_message_callback=callback, auto_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

其他说明

rabbitMQ有虚拟主机(virtual hosts)、交换机(Exchanges)、绑定(binding)、通道(Channels)、连接(Connections)、队列(Queues)。
- 虚拟主机为用户权限的最小单位，权限控制都是在虚拟主机这个层级进行控制的。
- 交换机控制消息的分发情况，可以群发到所有队列去，或者用绑定(binding)绑定到指定队列。默认为单一绑定，指定routing_key将消息发到对应队列去。
- 绑定为交换机的属性，告诉交换机怎么分发消息。
- 连接为物理连接，一个tcp连接一个连接。
- 通道为逻辑连接，一个物理连接可以有多个通道，即虚拟连接。不同线程/进程可以使用独立的通道，读写分离，更方便理解。
- 队列为最终的消息数组，消费者从某个队列直接消费数据。
默认消息是需要ack的，代表消息已经被处理完毕。上面例子中设置了auto_ack=True，所以不需要显性执行ack操作。正常而言，处理之后需要调用ch.basic_ack(delivery_tag=method.delivery_tag)来告诉rabbitMQ，这条消息已经被处理完毕。

多线程处理

pika是非多线程安全的！！！
所以如果想多线程处理消息，要么每个线程建立一个连接（Connections），要么使用python自带的queue来做缓冲。每个线程建立一个连接开销比较大，但是代码简单，在压力不大的情况下足够用了。使用内置队列，代码稍微复杂一点，但是性能更好。以下为内置队列的demo。

import pika,functools,queue
connection = pika.BlockingConnection(pika.ConnectionParameters(credentials=pika.PlainCredentials('small','123456')))
channel = connection.channel()
channel.queue_declare(queue='test')
class worker():
    q = queue.Queue(10)

    @staticmethod
    def work():
        while True:
            data = worker.q.get()
            body=data['body']
            # do something
            cutil.rabbit.add_callback_threadsafe(functools.partial(channel.basic_ack, data['tag']))

def rec(ch, method, properties, body):
    try:
        body = json.loads(body)
        worker.q.put({'body': body, 'tag': method.delivery_tag})
    except Exception as e:
        print('sender ???',end='')
        print(e,flush=True)

for x in range(10):
    threading.Thread(target=worker.work, daemon=True).start()
init()
channel.basic_consume(queue=RABBITC.SEND.value, on_message_callback=rec)
channel.start_consuming()

注意：在子线程不能直接调用channel.basic_ack，必须使用add_callback_threadsafe间接执行。更普遍的，在子线程不能直接调用channel的其他任何函数，都需要间接执行，add_callback_threadsafe会让channel在主线程执行对应函数。

心跳机制

在连接的时候，只要ConnectionParameters中的heartbeat不为0,rabbitMQ就会启用心跳机制，在超时没有数据且没有收到心跳包的情况下断开连接。
作为客户端（调用了start_consuming方法），pika在没有接收数据的时候，间隔一段时间会主动发送心跳包，维持连接。然而，作为服务端，pika没有事件时是不会发送心跳包的。这样会引起一个问题：当业务量很少，比如间隔几分钟才添加一条消息，pika却不会主动发心跳包。在中间等待数据的间隔，rabbitMQ会超时断开连接，pika再尝试添加消息时会出错。
解决方案有两种，一种是重连机制，发送失败就重新连接然后再发送。另外一种是手动发送心跳包，自己设置一个定时器，到时间了执行channel.queue_declare命令，pika就会正常发心跳包了。

rabbitMQ玩耍日记

其他说明

多线程处理

心跳机制

猜你喜欢

热点阅读