关于rabbitmq:RabbitMQ实践code

38次阅读

共计 8240 个字符，预计需要花费 21 分钟才能阅读完成。

简略模式 - 生产与生产

producer 生产者

import pika
import json
 
credentials = pika.PlainCredentials('guest', 'guest')
conn = pika.BlockingConnection(pika.ConnectionParameters(host='your robbitmq server', port=5672, virtual_host='/', credentials=credentials))
 
channel = conn.channel()
result = channel.queue_declare(queue='zy_test')
 
for i in range(100):
    message = json.dumps({'OrderId': f'1000{i}'})
    channel.basic_publish(exchange='', routing_key='zy_test', body=message)
    print(message)
 
conn.close()

consumer 消费者

import pika
import json
 
credentials = pika.PlainCredentials('guest', 'guest')
conn = pika.BlockingConnection(pika.ConnectionParameters(host='your robbitmq server', port=5672, virtual_host='/', credentials=credentials))
 
channel = conn.channel()
# durable 长久化 音讯不失落
channel.queue_declare(queue='zy_test', durable=True)
# 负载平衡：应用 basicQos(prefetchCount = 1)办法，来限度 RabbitMQ 只发不超过 1 条的音讯给同一个消费者。当音讯处理完毕后，有了反馈，才会进行第二次发送。channel.basic_qos(prefetch_count=1)
 
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(body.decode())
 
channel.basic_consume(queue='zy_test', on_message_callback=callback)
channel.start_consuming()

公布与订阅

公布与订阅要借助交换机（EXchange）的原理来实现：

EXchange 一共有四种工作模式：fanout，direct，topic，headers（不罕用）

1、fanout 模式

fanout 模式下，传递到 exchange 的音讯将会转发到所有与其 绑定的 queue上。艰深了解为：发送给与 exchange 绑定的所有 queue

  • 不须要指定 routing_key，即便指定了也是有效的
  • 须要提前将 exchange 和 queue 绑定，多对多关系：一个 exchange 能够绑定多个 queue，一个 queue 能够绑定多个 exchange
  • 须要先启动订阅者，此模式下的队列是 consumer 随机生成的，发布者仅仅发消息到 exchange，由 exchange 转发音讯至 queue。

发布者：

import json
from time import sleep
 
import pika
 
credentials = pika.PlainCredentials('guest', 'guest')
conn = pika.BlockingConnection(pika.ConnectionParameters(host='your robbitmq server', port=5672, virtual_host='/', credentials=credentials))
 
channel = conn.channel()
# 申明 exchange，由 exchange 指定音讯在哪个队列传递，如不存在，则创立。durable = True 代表 exchange 长久化存储，False 非长久化存储
channel.exchange_declare(exchange='zy_test', exchange_type='fanout', durable=True)
 
for i in range(100):
    message = json.dumps({'OrderId': f'20000{i}'})
    # 向队列插入数值 routing_key 是队列名。delivery_mode = 2 申明音讯在队列中长久化，delivery_mod = 1 音讯非长久化。routing_key 不须要配置
    channel.basic_publish(
        exchange='zy_test',
        routing_key='',
        body=message,
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f'Sent: {message}')
    sleep(1)
conn.close()

订阅者：

import pika
 
credentials = pika.PlainCredentials('guest', 'guest')
conn = pika.BlockingConnection(pika.ConnectionParameters(host='your robbitmq server', port=5672, virtual_host='/', credentials=credentials))
 
channel = conn.channel()
channel.exchange_declare(exchange='logs', exchange_type='fanout')
# 创立长期队列, 队列名传空字符，consumer 敞开后，队列主动删除
result = channel.queue_declare(queue='', exclusive=True)
# 绑定 exchange 和队列 queue，exchange 使咱们可能确切地指定音讯应该到哪个队列去
channel.queue_bind(exchange='zy_test', queue=result.method.queue)
 
 
# 回调函数，解决音讯内容
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(body.decode())
 
 
# auto_ack 设置成 False，在调用 callback 函数时，未收到确认标识，音讯会重回队列。True，无论调用 callback 胜利与否，音讯都被生产掉
channel.basic_consume(result.method.queue, callback, auto_ack=False)
channel.start_consuming()

2、direct 模式

direct 模式原理：音讯发送至 exchange，exchange 依据路由键（routing_key）转达到对应的 queue 上；艰深的了解为：依据 routing_key 过滤。

  • 能够应用默认 exchange=”，也能够自定义 exchange
  • direct 模式不须要将 exchange 和 queue，routing_key，queue 进行绑定，当然能够绑定；
  • 传递或接管音讯时 须要指定 routing_key
  • 须要先启动 订阅者，此模式下的 consumer 是随机生成的，发布者仅仅公布音讯到 exchange，由 exchange 转发音讯至 queue。如果后启动订阅者，则会失落启动前的音讯数据。

发布者：

import json
from time import sleep
 
import pika
 
credentials = pika.PlainCredentials('guest', 'guest')
conn = pika.BlockingConnection(pika.ConnectionParameters(host='your robbitmq server', port=5672, virtual_host='/', credentials=credentials))
 
channel = conn.channel()
channel.exchange_declare(exchange='zy_test_d', durable=True, exchange_type='direct')
for i in range(100):
    message = json.dumps({'OrderId': f'30000{i}'})
    # 向队列插入数值 routing_key 是队列名。delivery_mode = 2 申明音讯在队列中长久化，delivery_mod = 1 音讯非长久化。routing_key 不须要配置
    channel.basic_publish(
        exchange='zy_test_d',
        routing_key='OrderId',
        body=message,
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f'Sent: {message}')
    sleep(1)
conn.close()

订阅者：

import pika
 
credentials = pika.PlainCredentials('guest', 'guest')
conn = pika.BlockingConnection(pika.ConnectionParameters(host='your robbitmq server', port=5672, virtual_host='/', credentials=credentials))
 
channel = conn.channel()
channel.exchange_declare(exchange='zy_test_d', exchange_type='direct', durable=True)
# 创立长期队列, 队列名传空字符，consumer 敞开后，队列主动删除
result = channel.queue_declare(queue='', exclusive=True)
# 绑定 exchange 和队列 queue，exchange 使咱们可能确切地指定音讯应该到哪个队列去
channel.queue_bind(exchange='zy_test_d', queue=result.method.queue, routing_key='OrderId')
 
 
# 回调函数，解决音讯内容
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(body.decode())
 
 
# auto_ack 设置成 False，在调用 callback 函数时，未收到确认标识，音讯会重回队列。True，无论调用 callback 胜利与否，音讯都被生产掉
channel.basic_consume(result.method.queue, callback, auto_ack=False)
channel.start_consuming()

3、topic 模式

topic 模式和第二种模式差不多，exchange 也是通过 路由键 routing_key 来转发音讯到指定的 queue。_不同点是 routing_key 应用正则表达式反对含糊匹配 _，但匹配规定又与惯例的正则表达式不同，比方“#”是匹配全副，“*”是匹配一个词。
举例：routing_key =“#orderid#”，意思是将音讯转发至所有 routing_key 蕴含“orderid”字符的队列中。代码和模式二 相似，就不贴出来了。

基于 rabbitMQ 的 RPC

Callback queue 回调队列

一个客户端向服务器发送申请，服务器端解决申请后，将其处理结果保留在一个存储体中。而客户端为了取得处理结果，那么客户在向服务器发送申请时，同时发送一个回调队列地址 reply_to。

Correlation id 关联标识

一个客户端可能会发送多个申请给服务器，当服务器解决完后，客户端无奈分别在回调队列中的响应具体和那个申请时对应的。为了解决这种状况，客户端在发送每个申请时，同时会附带一个独有 correlation_id 属性，这样客户端在回调队列中依据 correlation_id 字段的值就能够分辨此响应属于哪个申请。

客户端发送申请

客户端在发送 RPC 申请到 RPC 申请队列时，客户端至多发送带有 reply_to 以及 correlation_id 两个属性的信息

服务端工作流

期待承受客户端发来 RPC 申请，当申请呈现的时候，服务器从 RPC 申请队列中取出申请，进行解决后，将响应发送到 reply_to 指定的回调队列中

客户端承受处理结果

客户端期待回调队列中呈现响应，当响应呈现时，它会依据响应中 correlation_id 字段的值，将其返回给对应的利用

server 端

import pika
 
 
def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)
 
 
# 解决 RPC 申请队列中数据，并回调
def on_request(ch, method, properties, body):
    n = int(body)
    print(f'[recv] n={n}')
    response = fib(n)
 
    ch.basic_publish(exchange='',
                     routing_key=properties.reply_to,
                     properties=pika.BasicProperties(correlation_id=properties.correlation_id),
                     body=str(response)
                     )
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
# 鉴权
credentials = pika.PlainCredentials('guest', 'guest')
# 建设链接
conn = pika.BlockingConnection(pika.ConnectionParameters(host='your rabbitmq host', port=5672, virtual_host='/', credentials=credentials))
# 建设会话
channel = conn.channel()
# 申明 RPC 申请队列
channel.queue_declare(queue='rpc_queue')
# 负载平衡
channel.basic_qos(prefetch_count=1)
# 生产客户端的音讯，并申请回调办法 on_request
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)
channel.start_consuming()

client 端

import uuid
import pika
 
 
class fibRpcClient:
    def __init__(self):
        self.__credentials = pika.PlainCredentials('guest', 'guest')
        self.conn = pika.BlockingConnection(pika.ConnectionParameters(host='your rabbitmq server', port=5672, virtual_host='/', credentials=self.__credentials))
        # 定义回调音讯会话
        self.channel = self.conn.channel()
        result = self.channel.queue_declare(queue='rpc_back', exclusive=True)
        self.callback_queue = result.method.queue
        self.correlation_id = str(uuid.uuid4())
        self.response = None
        # 收到回调音讯，开始生产
        self.channel.basic_consume(
            on_message_callback=self.on_response,
            queue=self.callback_queue,
            auto_ack=False
        )
 
    # 收到音讯后 数据处理办法
    def on_response(self, ch, method, properties, body):
        if self.correlation_id == properties.correlation_id:
            self.response = body
 
    def call(self, n):
        # 发送音讯
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',
            body=str(n),
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.correlation_id
            )
        )
        print(f'[send] n={n}')
        # 没有收到相应，就始终非阻塞式的 start_consumer
        while self.response is None:
            self.conn.process_data_events()
        print(f'[recv] result is {int(self.response)}')
 
if __name__ == '__main__':
    fib = fibRpcClient()
    fib.call(17)

长久化

MQ 默认建设的是长期 queue 和 exchange，如果不申明长久化，一旦 rabbitmq 挂掉，queue、exchange 将会全副失落。所以咱们个别在创立 queue 或者 exchange 的时候会申明 长久化。

1、queue 申明长久化

申明音讯队列，音讯将在这个队列传递，如不存在，则创立。
durable = True 代表音讯队列长久化存储，False 非长久化存储

result = channel.queue_declare(queue = 'python-test',durable = True)

2、exchange 申明长久化

申明 exchange，由 exchange 指定音讯在哪个队列传递，如不存在，则创立。
durable = True 代表 exchange 长久化存储，False 非长久化存储

channel.exchange_declare(exchange = 'python-test', durable = True)

3、音讯长久化

尽管 exchange 和 queue 都申明了长久化，但如果音讯只存在内存里，rabbitmq 重启后，内存里的货色还是会失落。所以必须申明音讯也是长久化，从内存转存到硬盘。

向队列插入数值 routing_key 是队列名。
delivery_mode = 2 申明音讯在队列中长久化，delivery_mod = 1 音讯非长久化 

channel.basic_publish(exchange = '',routing_key ='python-test',body = message, properties=pika.BasicProperties(delivery_mode = 2))

4、acknowledgement 音讯不失落

消费者（consumer）调用 callback 函数时，会存在解决音讯失败的危险，如果解决失败，则音讯失落。然而也能够抉择消费者解决失败时，将音讯回退给 rabbitmq，从新再被消费者生产，这个时候须要设置确认标识。

auto_ack 设置成 False，在调用 callback 函数时，未收到确认标识，音讯会重回队列。
True，无论调用 callback 胜利与否，音讯都被生产掉 

channel.basic_consume(callback,queue = 'python-test', auto_ack= False)

正文完
rabbitmq
发表至： rabbitMq
2020-12-28
 0
Laravel-使用-RabbitMQ
关于rabbitmq:Windows下RabbitMQ环境搭建
关于rabbitmq:系统学习消息队列RabbitMQ的交换机
关于rabbitmq:EMQ-X-与-RabbitMQ-消息服务器-MQTT-性能对比下