RabbitMQ

AMQP

AMQP协议是一个高级抽象层消息通信协议，RabbitMQ是AMQP协议的实现。它主要包括以下组件：
1.Server(broker): 接受客户端连接，实现AMQP消息队列和路由功能的进程。

2.Virtual Host:其实是一个虚拟概念，类似于权限控制组，一个Virtual Host里面可以有若干个Exchange和Queue，但是权限控制的最小粒度是Virtual Host

3.Exchange:交换机，接受生产者发送的消息，并根据Binding规则将消息路由给服务器中的队列。ExchangeType决定了Exchange路由消息的行为，在RabbitMQ中，ExchangeType有direct、Fanout和Topic三种，不同类型的Exchange路由的行为是不一样的。
- Direct直连交换机
    特点：依据key进行投递
    例如绑定时设置了routing key为”hello”，那么客户端提交的消息，只有设置了key为”hello”的才会投递到队列。
- Topic主题交换机
    特点：对key模式匹配后进行投递，符号”#”匹配一个或多个词，符号”*”匹配一个词
    例如”abc.#”匹配”abc.def.ghi”，”abc.*”只匹配”abc.def”。
- Fanout扇型交换机
    特点：不需要key，采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列

4.Message Queue：消息队列，用于存储还未被消费者消费的消息。

5.Message: 由Header和Body组成，Header是由生产者添加的各种属性的集合，包括Message是否被持久化、由哪个Message Queue接受、优先级是多少等。而Body是真正需要传输的APP数据。

6.Binding:Binding联系了Exchange与Message Queue。Exchange在与多个Message Queue发生Binding后会生成一张路由表，路由表中存储着Message Queue所需消息的限制条件即Binding Key。当Exchange收到Message时会解析其Header得到Routing Key，Exchange根据Routing Key与Exchange Type将Message路由到Message Queue。Binding Key由Consumer在Binding Exchange与Message Queue时指定，而Routing Key由Producer发送Message时指定，两者的匹配方式由Exchange Type决定。 

7.Connection:连接，对于RabbitMQ而言，其实就是一个位于客户端和Broker之间的TCP连接。

8.Channel:信道，仅仅创建了客户端到Broker之间的连接后，客户端还是不能发送消息的。需要为每一个Connection创建Channel，AMQP协议规定只有通过Channel才能执行AMQP的命令。一个Connection可以包含多个Channel。之所以需要Channel，是因为TCP连接的建立和释放都是十分昂贵的，如果一个客户端每一个线程都需要与Broker交互，如果每一个线程都建立一个TCP连接，暂且不考虑TCP连接是否浪费，就算操作系统也无法承受每秒建立如此多的TCP连接。RabbitMQ建议客户端线程之间不要共用Channel，至少要保证共用Channel的线程发送消息必须是串行的，但是建议尽量共用Connection。

9.Command:AMQP的命令，客户端通过Command完成与AMQP服务器的交互来实现自身的逻辑。例如在RabbitMQ中，客户端可以通过publish命令发送消息，txSelect开启一个事务，txCommit提交一个事务。

# rabbitmq官方推荐的python客户端pika模块
pip3 install pika

python sender.py First message.
python sender.py Second message..
python sender.py Third message...
python sender.py Fourth message....
python sender.py Fifth message.....

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

channel.basic_publish(
                exchange=‘‘,
                 routing_key="task_queue",
                 body=message,
                 # delivery_mode = 2 使消息持久化
                 properties=pika.BasicProperties(delivery_mode = 2,)
             )

channel.basic_qos(prefetch_count=1)

channel.exchange_declare(exchange=‘logs‘,
                         exchange_type=‘fanout‘)

channel.basic_publish(exchange=‘‘,
                      routing_key=‘hello‘,
                      body=message)

channel.basic_publish(exchange=‘logs‘,
                      routing_key=‘‘,
                      body=message)

result = channel.queue_declare(queue="")

result.method.queue

result = channel.queue_declare(queue="",exclusive=True)  # 断开连接，队列立即删除

channel.queue_bind(exchange=‘logs‘,
                   queue=result.method.queue)

import pika
import sys
# 创建凭证，使用rabbitmq用户密码登录
credentials = pika.PlainCredentials("guest", "guest")
# 创建一个rabbitmq连接
connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘, credentials=credentials))
channel = connection.channel()
# 指定交换机类型
channel.exchange_declare(exchange=‘m1‘,exchange_type=‘fanout‘)

message = ‘ ‘.join(sys.argv[1:]) or "很高兴见到你"
# 指定发送给哪个交换机(exchange=‘‘)路由键(routing_key=‘‘),参数body值发送的数据
channel.basic_publish(
    exchange=‘m1‘,
    routing_key=‘‘,
    body=message,
)
print(" [x] Sent %r" % (message,))
connection.close()

import time
import pika

credentials = pika.PlainCredentials("guest", "guest")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘, credentials=credentials))
channel = connection.channel()
channel.exchange_declare(exchange=‘m1‘, exchange_type=‘fanout‘)
# 随机创建一个队列，扇形交换机要将消息给所有人，所以不需要特意指定一个队列的名称。
result = channel.queue_declare(queue="", exclusive=True)
# 获取队列的名字
queue_name = result.method.queue
# 创建绑定，交换机和队列
channel.queue_bind(exchange=‘m1‘,
                   queue=queue_name)

def callbak(ch, method, properties, body):
    print("[x] Received %r" % body.decode("utf8"))
    time.sleep(body.decode("utf8").count(‘.‘))
    print(" [x] Done")

channel.basic_consume(queue=queue_name, on_message_callback=callbak, auto_ack=True)
# 开始消费，接收消息
channel.start_consuming()

python receiver.py > logs_from_rabbit.log  # 使用该方式启动receiver.py

channel.queue_bind(exchange=exchange_name,
                   queue=queue_name,
                   routing_key=‘black‘)

import pika

routing_key=input("请输入routing_key:")
credentials = pika.PlainCredentials("guest", "guest")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘, credentials=credentials))
channel = connection.channel()
# 指定直连交换机
channel.exchange_declare(exchange=‘task1‘,exchange_type=‘direct‘)
message = f"很高兴见到你{routing_key}"
channel.basic_publish(
    exchange=‘task1‘,
    routing_key=routing_key,
    body=message,
)
print(f"给{routing_key}发送消息成功")
connection.close()

import time
import pika

credentials = pika.PlainCredentials("guest", "guest")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘, credentials=credentials))
channel = connection.channel()
channel.exchange_declare(exchange=‘task1‘, exchange_type=‘direct‘)
# 随机创建一个队列
result = channel.queue_declare(queue="", exclusive=True)
# 获取队列的名字
queue_name = result.method.queue
# 绑定交换机和队列
channel.queue_bind(exchange=‘task1‘,
                   queue=queue_name,
                   routing_key="xiaoqi")

def callbak(ch, method, properties, body):
    print(body.decode("utf8"))

channel.basic_consume(queue=queue_name, on_message_callback=callbak, auto_ack=True)
# 开始消费，接收消息
channel.start_consuming()

import time
import pika

credentials = pika.PlainCredentials("guest", "guest")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘, credentials=credentials))
channel = connection.channel()
channel.exchange_declare(exchange=‘task1‘, exchange_type=‘direct‘)
# 随机创建一个队列
result = channel.queue_declare(queue="", exclusive=True)
# 获取队列的名字
queue_name = result.method.queue
# 绑定交换机和队列
channel.queue_bind(exchange=‘task1‘,
                   queue=queue_name,
                   routing_key="dada")

def callbak(ch, method, properties, body):
    print(body.decode("utf8"))

channel.basic_consume(queue=queue_name, on_message_callback=callbak, auto_ack=True)
# 开始消费，接收消息
channel.start_consuming()

python receive_logs_direct.py warning error > logs_from_rabbit.log  # 此方式开启receiver

import pika

routing_key=input("请输入routing_key:")
credentials = pika.PlainCredentials("guest", "guest")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘, credentials=credentials))
channel = connection.channel()
# 指定直连交换机
channel.exchange_declare(exchange=‘task2‘,exchange_type=‘topic‘)
message = f"很高兴见到你{routing_key}"
channel.basic_publish(
    exchange=‘task2‘,
    routing_key=routing_key,
    body=message,
)
print(f"给{routing_key}发送消息成功")
connection.close()

import time
import pika

credentials = pika.PlainCredentials("guest", "guest")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘, credentials=credentials))
channel = connection.channel()
channel.exchange_declare(exchange=‘task2‘, exchange_type=‘topic‘)
# 随机创建一个队列
result = channel.queue_declare(queue="", exclusive=True)
# 获取队列的名字
queue_name = result.method.queue
# 绑定交换机和队列
channel.queue_bind(exchange=‘task2‘,
                   queue=queue_name,
                   routing_key="*.md")

def callbak(ch, method, properties, body):
    print(body.decode("utf8"))
# 有消息来临，立即执行callbak，从队列q1中取数据；没有消息则夯住，等待消息
channel.basic_consume(queue=queue_name, on_message_callback=callbak, auto_ack=True)
# 开始消费，接收消息
channel.start_consuming()

下单  { 库存>减少库存    
        支付>扣款    
        红包>减免红包    
        物流>生成订单}

result = channel.queue_declare(queue="",exclusive=True)
callback_queue = result.method.queue

channel.basic_publish(exchange=‘‘,
                      routing_key=‘rpc_queue‘,
                      properties=pika.BasicProperties(
                            reply_to = callback_queue,
                            ),
                      body=request)

import pika
import uuid

class FibonacciRpcClient(object):
    def __init__(self):
        credentials = pika.PlainCredentials("guest", "guest")
        # 客户端启动时，创建回调队列，会开启会话用于发送RPC请求以及接受响应
        # 建立连接，指定服务器的ip地址
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host=‘localhost‘, credentials=credentials))
        # 建立一个会话，每个channel代表一个会话任务
        self.channel = self.connection.channel()

        # 声明回调队列，再次声明的原因是，服务器和客户端可能先后开启，该声明是幂等的，多次声明，但只生效一次
        # exclusive=True 参数是指只对首次声明它的连接可见
        # exclusive=True 会在连接断开的时候，自动删除
        result = self.channel.queue_declare(queue="", exclusive=True)
        # 将次队列指定为当前客户端的回调队列，用户接收服务端给客户端发送的信息
        self.callback_queue = result.method.queue
        # 客户端订阅回调队列，当回调队列中有响应时，调用`on_response`方法对响应进行处理;
        # 有消息来临，立即执行callbak，从队列on_response中取数据；没有消息则夯住，等待消息
        self.channel.basic_consume(queue=self.callback_queue, on_message_callback=self.on_response, auto_ack=True)

    # 对回调队列中的响应进行处理的函数
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    # 发出RPC请求
    # 例如这里服务端就是一场马拉松，跑完一段将棒交给下一个人，这个过程是发送rpc请求
    def call(self, n):
        # 初始化 response
        self.response = None
        # 生成correlation_id 关联标识，通过python的uuid库，生成全局唯一标识ID，保证时间空间唯一性
        self.corr_id = str(uuid.uuid4())
        # 发送RPC请求内容到RPC请求队列`task2`，同时发送的还有`reply_to`和`correlation_id`
        self.channel.basic_publish(exchange=‘‘,
                                   routing_key=‘main_queue‘,
                                   properties=pika.BasicProperties(
                                       reply_to=self.callback_queue,
                                       correlation_id=self.corr_id,
                                   ),
                                   body=str(n))
        while self.response is None:
            self.connection.process_data_events()
        return int(self.response)

# 建立客户端
fibonacci_rpc = FibonacciRpcClient()

# 发送RPC请求，丢进rpc队列，等待客户端处理完毕，给与响应
print("发送了请求sum(100)")
response = fibonacci_rpc.call(100)

print("得到远程结果响应%r" % response)

import pika

credentials = pika.PlainCredentials("guest", "guest")
# 建立连接，服务器地址为localhost，也可指定ip地址
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘,credentials=credentials))
# 建立会话
channel = connection.channel()
# 模拟一个进程，例如接力比赛
def sum(n):
    n+=100
    return n
# 对RPC请求队列中的请求进行处理

def on_request(ch, method, props, body):
    print(body,type(body))
    n = int(body)
    print(" 正在处理sum(%s)..." % n)
    # 调用数据处理方法
    response = sum(n)
    # 将处理结果(响应)发送到回调队列
    ch.basic_publish(exchange=‘‘,
                     # reply_to代表回复目标
                     routing_key=props.reply_to,
                     # correlation_id（关联标识）：用来将RPC的响应和请求关联起来。
                     properties=pika.BasicProperties(correlation_id= props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(" 处理完成sum(%s)" % n)

# 负载均衡，同一时刻发送给该服务器的请求不超过一个
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue=‘main_queue‘,on_message_callback=on_request)
print("等待接收rpc请求")

#开始消费
channel.start_consuming()