Rabbitmq-server 分析

Posted 2020-11-09

一、
发送端：

#coding:utf-8
import pika

user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))     #建立一个实例
channel = connection.channel()  # 声明一个管道，在管道里发消息
channel.queue_declare(queue=‘hello‘)  #在管道里声明queue，并指定queue_name
#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‘,  # queue名字
                      body=‘Hello World! ‘)  # 消息内容
print(" [x] Sent ‘Hello World!‘")
connection.close()  # 队列关闭

接收端：

#coding:utf-8
import pika
import time

user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.206‘, credentials=user_pwd))             # 建立实例
channel = connection.channel()   # 声明管道

# 在不确定procedure与consumer哪个先运行的情况下，在consumer端 再次声明QUEUE。
channel.queue_declare(queue=‘hello‘)

def callback(ch, method, properties, body):  #定义回调函数，四个参数为标准格式
    print(" [x] Received %r" % body)    # 输出接收到的消息
    time.sleep(15)
    ch.basic_ack(delivery_tag=method.delivery_tag)  # 告诉生产者消息处理完成

channel.basic_consume(  # 消费消息
        callback,  # 如果收到消息，就调用callback函数来处理消息
        queue=‘hello‘,  # 你要从那个队列里收消息
        #no_ack=True   # True 表示consumer取完消息后不给procedure发送ack确认，注意本例中如果为True，它会与callback函数中的basic_ack语句产生冲突；默认False
        )

print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
channel.start_consuming()  # 开始消费消息

验证：
在多个shell窗口中运行consumer，然后运行procedure，发现每个consumer 轮流接收相同队列中的消息。
二、
发送端：

#coding:utf-8
import pika

user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
channel = connection.channel()

channel.queue_declare(queue=‘hello2‘, durable=True)  #创建一个新队列task_queue，设置队列持久化，注意不要跟已存在的队列重名，否则有报错
#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=‘hello2‘, body=‘Hello World 2 !‘,
                      properties=pika.BasicProperties(delivery_mode=2, )   # make message persistent
                      )

print(" [x] Sent ‘Hello World!‘")
connection.close()

接收端：

#coding:utf-8
import pika
import time

user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
channel = connection.channel()
channel.queue_declare(queue=‘hello2‘, durable=True)

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    time.sleep(10)
    ch.basic_ack(delivery_tag=method.delivery_tag)  # 告诉生产者，消息处理完成

channel.basic_qos(prefetch_count=1)  # 类似权重，按能力分发，如果有一个消息，就不在给你发
channel.basic_consume(  # 消费消息
                      callback,  # 如果收到消息，就调用callback
                      queue=‘hello2‘,
                      )
print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
channel.start_consuming()

(1) rabbitmq循环调度，将消息循环发送给不同的消费者
(2) 消息确认机制，为了确保一个消息不会丢失，RabbitMQ支持消息的确认 , 一个 ack(acknowlegement) 是从消费者端发送一个确认去告诉RabbitMQ 消息已经接收了、处理了，RabbitMQ可以释放并删除掉了。如果一个消费者死掉了（channel关闭、connection关闭、或者TCP连接断开了）而没有发送ack，RabbitMQ 就会认为这个消息没有被消费者处理，并会重新发送到生产者的队列里，如果同时有另外一个消费者在线，rabbitmq将会将消息很快转发到另外一个消费者中。
(3) 消息持久化，将消息写入硬盘中。RabbitMQ不允许你重新定义一个已经存在、但属性不同的queue。关于消息为持久化， 需要生产者定义持久化的queue，以及通过设置 delivery_mode 属性为 2来标记本消息持久化。
(4) 公平调度。在一个消费者未处理完一个消息之前不分发新的消息给它，而是将这个新消息分发给另一个不是很忙的消费者进行处理。为了解决这个问题可以在消费者代码中使用 channel.basic.qos ( prefetch_count = 1 )，将消费者设置为公平调度
验证：
在多个shell窗口中打开consumer，然后运行procedure，观察收到消息为RR

三、
exchange：交换机。生产者不是将消息发送给队列，而是将消息发送给交换机，由交换机决定将消息发送给哪个队列。所以exchange必须准确知道消息是要送到哪个队列，还是要被丢弃。因此要在exchange中给exchange定义规则，所有的规则都是在exchange的类型中定义的。
(1) fanout
生产者：

#coding:utf-8
import pika
import sys

user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
#建立一个实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
channel = connection.channel()

# 注意：这里是广播，不需要声明queue
channel.exchange_declare(‘logs‘, ‘fanout‘)
message = ‘ ‘.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange=‘logs‘,
                      routing_key=‘‘,  # 注意此处空，必须有
                      body=message)
print(" [x] Sent %r" % message)
connection.close()

消费者：

#coding:utf-8
import pika
import time

user_pwd = pika.PlainCredentials(‘admin‘,‘admin‘)
# 建立实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
# 声明管道
channel = connection.channel()
channel.exchange_declare(‘logs‘, ‘fanout‘)

# 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
result = channel.queue_declare(exclusive=True)
# 获取随机的queue名字
queue_name = result.method.queue
#print("random queuename:", queue_name)

channel.queue_bind(exchange=‘logs‘,  # queue绑定到转发器上
                   queue=queue_name)
print(‘ [*] Waiting for logs. To exit press CTRL+C‘)

def callback(ch, method, properties, body):
    print(" [x] %r" % body)

channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()

# 注意：广播是实时的，收不到就没了，消息不会存下来，类似收音机

(2) direct
生产者：

#coding:utf-8
import pika
import sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
#建立一个实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
channel = connection.channel()
channel.exchange_declare(‘direct_logs‘, ‘direct‘)

# 重要程度级别，这里默认定义为 info
severity = sys.argv[1] if len(sys.argv) > 1 else ‘warning‘

message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World! direct log‘
channel.basic_publish(exchange=‘direct_logs‘, routing_key=severity, body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

消费者：

#coding:utf-8
import pika
import time, sys

user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
# 建立实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
channel = connection.channel()

channel.exchange_declare(‘direct_logs‘, ‘direct‘)
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

# 获取运行脚本所有的参数
severities = sys.argv[1:]
if not severities:
    #severities = [‘error‘, ‘warning‘]
    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
    sys.exit(1)

# 循环列表去绑定
for severity in severities:
    channel.queue_bind(exchange=‘direct_logs‘, queue=queue_name, routing_key=severity)

print(‘ [*] Waiting for logs. To exit press CTRL+C‘)

def callback(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))

channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()

(3) topic
生产者:

#coding:utf-8
import pika
import sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
#建立一个实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
# 声明一个管道，在管道里发消息
channel = connection.channel()

channel.exchange_declare(‘topic_logs‘, ‘topic‘)

routing_key = sys.argv[1] if len(sys.argv) > 1 else ‘anonymous.info‘
message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘
channel.basic_publish(exchange=‘topic_logs‘, routing_key=routing_key, body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()

消费者：

#coding:utf-8
import pika
import time, sys

user_pwd = pika.PlainCredentials(‘admin‘,‘admin‘)
# 建立实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
channel = connection.channel()

channel.exchange_declare(‘topic_logs‘, ‘topic‘)
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
    sys.exit(1)

for binding_key in binding_keys:
    channel.queue_bind(exchange=‘topic_logs‘, queue=queue_name, routing_key=binding_key)

print(‘ [*] Waiting for logs. To exit press CTRL+C‘)

def callback(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))

channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()