RabbitMQ消息队列

一、消息队列

消息队列就是一种先进先出的数据机构

当在分布式系统中的时候，不同的机器需要做数据交互，所以涉及到不同机器之间的数据交互，这样的话就需要借助专业的消息队列，常见的消息队列有 RabbitMQ 、Kafka...他们都是开源且支持语言较多。

消息队列解决的问题：

1. 应用解耦

2. 流量消峰：

   如果订单系统一秒最多能处理一万次订单，这个处理能力在平时绰绰有余，正常时段我们下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的，只能限制订单超过一万后不允许用户下单。

   但是使用消息队列，就可以取消这个限制，把这一秒内的订单放入队列中分散成一段时间来处理，这样用户就可能在下单几十秒后才能收到下单成功的操作，但是比不能下单要好。

3. 消息分发：

   当A发送一次消息，B对消息感兴趣，就只需监听消息，C感兴趣，C也去监听消息，而A完全不需要改动。

4. 异步消息（Celery 就是对消息队列的分装）

RabbitMQ 和 Kafka

RabbitMQ ：吞吐量小，有消息确认（对消息可靠性有要求，就用它）

Kafka：吞吐量高，注重高吞吐量，不注重消息的可靠性，数据量特别大

二、按装RabbitMQ

1、原生安装：

# 安装扩展epel源
wget -O /etc/yum.repos.d/epel.repo http://mirrors.aliyun.com/repo/epel-7.repo
    
yum -y install erlang			# 因为RabbitMQ是erlang语言开发的，所以要按装
yum -y install rabbitmq-server	# 安装RabbitMQ
systemctl start rabbitmq-server	# 启动

# 创建用户
rabbitmqctl add_user 用户名 密码
# 分配权限
rabbitmqctl set_user_tags 用户名 administrator ——>(设置用户为管理员角色)
rabbitmqctl set_permissions -p "/" 用户名 ".*" ".*" ".*"	# 设置权限
systemctl reatart rabbitmq-server	# 重启

2、docker拉取

docker pull rabbitmq:3.8.3-management		# 自动开启了web管理界面

# 启动需要配置用户名和密码
docker run -di --name rabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 rabbitmq:3.8.3-management

5672：是RabbitMQ的默认端口
15672：web管理界面的端口

三、基本使用

生产者：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))

# 拿到channel对象
channel = connection.channel()

# 声明一个队列
channel.queue_declare(queue='test')

# 生产者向队列中放入一条消息
channel.basic_publish(exchange='',
                      routing_key='test',   # 指定向那个队列放入
                      body='测试数据')        # 放入的内容
                      
# 关闭连接
connection.close()

消费者：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))

# 拿到channel对象
channel = connection.channel()

# 声明一个队列，如果消费者先起来，那么就先声明一个队列
channel.queue_declare(queue='test')


def callback(ch, method, properties, body):
    print(f'测试：{body}')


# 消费者从指定的队列中拿消息消费，一旦有一条转到 callback 里
channel.basic_consume(queue='test', on_message_callback=callback, auto_ack=True)

# 阻塞主，一直等待拿消息消费
channel.start_consuming()

四、确认机制

消息确认机制其实就是消费者中 auto_ack 的设置

生产者不变

消费者：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))

# 拿到channel对象
channel = connection.channel()

# 声明一个队列，如果消费者先起来，那么就先声明一个队列
channel.queue_declare(queue='test')


def callback(ch, method, properties, body):
    print(f'测试：{body}')
    # 如果auto_ack=False这样设置后
    # 也可以这样设置当真正的消息处理完了，在发确认也是可以的
    ch.basic_ack(delivery_tag=method.delivery_tag)


# auto_ack=True，队列收到确认，就会自动把消费过的消息删除。
# auto_ack=False，那么就不会给队列发送确认消息了，队列就不会删除消息。不会自动回复确认消息，
channel.basic_consume(queue='test', on_message_callback=callback, auto_ack=False)

# 阻塞主，一直等待拿消息消费
channel.start_consuming()

五、持久化

队列持久化：就是在声明队列的时候，指定持久化durable=True，队列必须是新的才可以

channel.queue_declare(queue='test', durable=True)	# test 队列持久化

消息持久化：就是在发布消息的时候添加

# 生产者向队列中放入一条消息
channel.basic_publish(exchange='',
                      routing_key='test',   # 指定向那个队列放入
                      body='hello',         # 放入的内容
                      properties=pika.BasicProperties(delivery_mode=2) # 消息持久化
                      )

六、闲置消费

当正常情况下如果有多个消费者，那么就会按照顺序第一个消息给 第一个消费者，第二个消息给第二个消费者

但是当第一个消息的消费者处理信息很耗时，一直没有结束，那么就可以让第二个消费者优先获取闲置消息。

消费者：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))

# 拿到channel对象
channel = connection.channel()

# 声明一个队列，如果消费者先起来，那么就先声明一个队列
channel.queue_declare(queue='test')


def callback(ch, method, properties, body):
    print(f'测试：{body}')


# 就只有这一句话，谁闲置谁获取，没必要按照顺序一个一个来
channel.basic_qos(prefetch_count=1)

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

# 阻塞主，一直等待拿消息消费
channel.start_consuming()

七、发布订阅

发布订阅就是：我可以有多个订阅者来订阅你的消息，这样发布者只需要发布一条， 我的所有只要订阅你的人都可以消费你的消息

模型：当我的订阅者起来了之后，就会创建一个队列，多个订阅者就会创建多个队列，当发布者生产了消息之后，会传给 exchange ，然后 exchange 会把消息复制分别分发到订阅者创建的队列中，这样就实现了只要监听你，那就能收到你发的消息。

基本使用

发布者：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))

# 拿到channel对象
channel = connection.channel()

# 不指定队列，指定了 exchange 复制分发消息
channel.exchange_declare(exchange='conn', exchange_type='fanout')

# 生产者向队列中放入一条消息
channel.basic_publish(exchange='conn',      # 指定复制分发消息的 exchange
                      routing_key='',   	# 不设置指定向那个队列放入
                      body='Hello Word',    # 放入的内容
                      )
# 关闭连接
connection.close()

订阅者：启动多次，都绑定到了同一个 exchange，所以就会都收到同一个 exchange 分发的消息

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))


# 拿到channel对象
channel = connection.channel()


# 声明一个队列，如果消费者先起来，那么就先声明一个队列
channel.exchange_declare(exchange='conn', exchange_type='fanout')


# queue 不能制定名字，因为它们的名字都是不一样的
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue    # 生成一个随机的 queue 名字



# 把随机生成的队列绑定到exchange上
channel.queue_bind(exchange='conn', queue=queue_name)


def callback(ch, method, properties, body):
    print(f'测试：{body}')


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

# 阻塞主，一直等待拿消息消费
channel.start_consuming()

关键字

需要设置 exchange_type 的类型为 direct

并且在发布消息的时候设置多个关键字：routing_key

在订阅者中也需要设置 exchange_type 的类型 direct

并且当订阅者绑定 exchange 的时候也需要设置 routing_key，

这样的话在发布者发布消息后，exchange 会根据发布者和订阅者设置的 routing_key 进行匹配，当订阅者的 routing_key 匹配上了发布者的 routing_key 的话，那么订阅者就可以接收到发布者发布的消息，反之收不到消息。

发布者：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))

# 拿到channel对象
channel = connection.channel()

# 不指定队列，指定了 exchange 复制分发消息，exchange_type='direct'
channel.exchange_declare(exchange='conn1', exchange_type='direct')

# 生产者向队列中放入一条消息
channel.basic_publish(exchange='conn1',     # 指定复制分发消息的 exchange
                      routing_key='abc',   # 指定关键字
                      body='Hello Word',    # 放入的内容
                      )
# 关闭连接
connection.close()

消费者1：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))


# 拿到channel对象
channel = connection.channel()


# 声明一个队列，如果消费者先起来，那么就先声明一个队列
channel.exchange_declare(exchange='conn', exchange_type='direct')


# queue 不能制定名字，因为它的名字都是不一样的
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue    # 生成一个随机的 queue 名字
print(queue_name)


# 把随机生成的队列绑定到exchange上，
# 并设置routing_key='abc'，也就是说只有发布者的routing_key中包含有'abc'，此订阅者才会收到消息
channel.queue_bind(exchange='conn1', queue=queue_name, routing_key='abc')


def callback(ch, method, properties, body):
    print(f'测试：{body}')


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

# 阻塞主，一直等待拿消息消费
channel.start_consuming()

消费者2：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))


# 拿到channel对象
channel = connection.channel()


# 声明一个队列，如果消费者先起来，那么就先声明一个队列
channel.exchange_declare(exchange='conn', exchange_type='direct')


# queue 不能制定名字，因为它的名字都是不一样的
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue    # 生成一个随机的 queue 名字
print(queue_name)


# 把随机生成的队列绑定到exchange上，
# 并设置了多个routing_key，也就是说只有发布者的routing_key中包含有入下两个之一，此订阅者都会收到消息
channel.queue_bind(exchange='conn1', queue=queue_name, routing_key='abc')
channel.queue_bind(exchange='conn1', queue=queue_name, routing_key='abcd')


def callback(ch, method, properties, body):
    print(f'测试：{body}')


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

# 阻塞主，一直等待拿消息消费
channel.start_consuming()

模糊匹配

在订阅者绑定匹配的时候可以进行模糊匹配发布者的 routing_key ，匹配上了就能接收到发布者发布的消息

# 表示后面可以跟任意字符
* 表示后面只能跟一个单词

发布者：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))

# 拿到channel对象
channel = connection.channel()

# 不指定队列，指定了 exchange 复制分发消息，exchange_type='topic'
channel.exchange_declare(exchange='conn1', exchange_type='topic')

# 生产者向队列中放入一条消息
channel.basic_publish(exchange='conn2',     # 指定复制分发消息的 exchange
                      routing_key='abcdefg',    # 指定关键字
                      body='Hello Word',    # 放入的内容
                      )
# 关闭连接
connection.close()

订阅者：

import pika

# 有用户名密码
credentials = pika.PlainCredentials('admin', 'admin')
# 拿到连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.88.131', credentials=credentials))


# 拿到channel对象
channel = connection.channel()


# 声明一个队列，如果消费者先起来，那么就先声明一个队列
channel.exchange_declare(exchange='conn', exchange_type='direct')


# queue 不能制定名字，因为它的名字都是不一样的
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue    # 生成一个随机的 queue 名字
print(queue_name)


# 把随机生成的队列绑定到exchange上，
# 并设置routing_key='abc#'，也就是说只有发布者的routing_key中包含有'abc'开头，此订阅者才会收到消息
channel.queue_bind(exchange='conn2', queue=queue_name, routing_key='abc#')


def callback(ch, method, properties, body):
    print(f'测试：{body}')


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

# 阻塞主，一直等待拿消息消费
channel.start_consuming()

八、python中的RPC框架

RPC ：远程过程调用

例如：两个服务调用，服务1通过网络调用服务2的方法。

SimpleXMLRPCServer

自带的：数据包大，速度慢

服务端：

from xmlrpc.server import SimpleXMLRPCServer


class RPCServer(object):

    def getObj(self):
        return 'get obj'

    def sendObj(self, data):
        return 'send obj'


# SimpleXMLRPCServer
server = SimpleXMLRPCServer(('localhost', 4242), allow_none=True)
server.register_introspection_functions()
server.register_instance(RPCServer())
server.serve_forever()

客户端：

from xmlrpc.client import ServerProxy

client = ServerProxy('http://localhost:4242')
ret = client.getObj()
print(ret)

ZeroRPC

第三方的：底层使用 ZeroMQ 和 MessagePack ，速度快，响应时间短，并发高。

服务端：

import zerorpc

class RPCServer(object):

    def getObj(self):
        return 'get obj'

    def sendObj(self, data):
        return 'send obj'


server = zerorpc.Server(RPCServer())
server.bind('tcp://0.0.0.0:4243')   # 允许连接的
server.run()

客户端：

import zerorpc

client = zerorpc.Client()
client.connect('tcp://127.0.0.1:4243')  # 连接
ret = client.getObj()
print(ret)