python操作rabbitmq

一、python简单rabbitmq消息队列(消息不安全,不持久化)

pip install pika

1)发送端

import pika

# rabbitmqctl set_permissions -p / rabbitadmin ".*" ".*" ".*"           注意创建用户后,一定要授权
credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))
channel = connection.channel()   # 建立了rabbit协议的通道

#声明queue
channel.queue_declare(queue='hello')

#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='hello',
                      body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()
send.py

2)接收端

import pika


credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))
channel = connection.channel()


#You may ask why we declare the queue again ‒ we have already declared it in our previous code.
# We could avoid that if we were sure that the queue already exists. For example if send.py program
#was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
channel.queue_declare(queue='hello')

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

channel.basic_consume(callback,
                      queue='hello',
                      no_ack=True)

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

接收到处于死循环,一直在等待接收,发送一个数据,就收到一个数据

 3)对接收端的callback函数内容详解

def callback(ch, method, properties, body):
    print(" [x] Received %r" % ch, method, properties, body)
callback函数详解

 二、消息队列深入理解

1)当有一个生产者,2个消费者时

基于上面的代码不做任何修改

运行2个接收者,生产者生成的消息队列依次被接收者接收

 2)处理消息安全问题(缺持久化)

基于上面代码,如果消费者出问题了,消息发送将无人接收。即便再次启动消费者,之前发生的消息将一直存在队列中

2.1生产者代码

import pika
import time
credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))
channel = connection.channel()

channel.queue_declare(queue='task_queue')

import sys

message = ' '.join(sys.argv[1:]) or "Hello World! %s" % time.time()
channel.basic_publish(exchange='',
                      routing_key='task_queue',
                      body=message,
                      # properties=pika.BasicProperties(
                      #     delivery_mode=2,  # make message persistent
                      # )
                      )
print(" [x] Sent %r" % message)
connection.close()
send_msg_safe.py

2.2)消费者代码

import pika, time

credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))
channel = connection.channel()


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    time.sleep(20)
    print(" [x] Done")
    print("method.delivery_tag",method.delivery_tag)
    ch.basic_ack(delivery_tag=method.delivery_tag)  # 再进行手动确认


channel.basic_consume(callback,
                      queue='task_queue',
                      # no_ack=True ,是需要是否确定消息的处理了,告诉服务端
                      )

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

问题出在于消费者:消费者处理好的消息,需要给服务端回信息

 # no_ack=True ,是需要是否确定消息的处理了,告诉服务端
 # no_ack=False ,默认是False,可以不写
 # callback 函数后面需要添加 ch.basic_ack(delivery_tag=method.delivery_tag)  # 再进行手动确认

 3)基于上面的代码,如果重启了rabbitmq,则存在的消息就消失。需要做到消息持久化。(消息安全且持久化)

3.1)生产者代码

import pika

# 通过实例创建socket
credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))

channel = connection.channel()

channel.queue_declare(queue='hello10',durable=True)

channel.basic_publish(exchange='',
                      routing_key='hello10',
                      body='Hello World!',
                      properties = pika.BasicProperties(
                      delivery_mode = 2,)
                      )
print(" [x] Sent 'Hello World!'")

# 关闭队列
connection.close()
send_msg.py

3.2) 消费者代码

import pika,time


credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))

channel = connection.channel()

channel.queue_declare(queue='hello10',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_consume(
                    callback,
                    queue='hello10',)

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

channel.start_consuming()
recv_msg.py

问题在于生产者的消息需要被持久化

channel.queue_declare(queue='hello10',durable=True)

properties = pika.BasicProperties(
delivery_mode = 2,)
)

三、消息的发布,订阅。广播模式

之前的例子都基本都是1对1的消息发送和接收,即消息只能发送到指定的queue里,
但有些时候你想让你的消息被所有的Queue收到,类似广播的效果,这时候就要用到exchange了,

Exchange在定义的时候是有类型的,以决定到底是哪些Queue符合条件,可以接收消息
fanout: 所有bind到此exchange的queue都可以接收消息
direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息
topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息
   表达式符号说明:#代表一个或多个字符,*代表任何字符
      例:#.a会匹配a.a,aa.a,aaa.a等
          *.a会匹配a.a,b.a,c.a等
     注:使用RoutingKey为#,Exchange Type为topic的时候相当于使用fanout 

1)广播模式,发送一个消息,无论有多少接收端,只要在,就能收到,不在就不能收到

1.1)发送端代码

import pika
import sys
credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))
channel = connection.channel()

channel.exchange_declare(exchange='logs',exchange_type='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()
send广播

1.2)接收端代码

import pika

credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))
channel = connection.channel()

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

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

channel.queue_bind(exchange='logs',
                   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()
recv广播

2)组播模式。

RabbitMQ还支持根据关键字发送,即:队列绑定关键字,发送者将数据根据关键字发送到消息exchange,exchange根据 关键字 判定应该将数据发送至指定队列。

send端
    可发送info,warning,error
    
recv端根据关键字指定接收内容
recv端1:
    接收error
recv端2:
    接收warning,error

2.1)发送端代码测试

import pika
import sys

credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',
                         exchange_type='direct')

severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()
send_组播.py

执行示例,需说明发送的内容级别

python send_组播.py info
python send_组播.py warning
python send_组播.py error

2.2)接收端代码测试

import pika
import sys

credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',
                         exchange_type='direct')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

severities = sys.argv[1:]
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]
" % 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()
recv_组播.py

执行示例,需说明接收代码内容

python recv组播.py info
python recv组播.py warning
python recv组播.py error

 3)基于组播模式,实现更细致的划分不同种类的信息

1)发送端

import pika
import sys

credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))
channel = connection.channel()

channel.exchange_declare(exchange='topic_logs',
                         exchange_type='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()


# python topic_send.py python.error test    发送了一条python的错误信息,错误内容为test
# python topic_send.py mysql.info hello     发送了一条mysql的信息,信息内容为hello
topic_send.py

2)接收端

import pika
import sys

credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
               '192.168.111.230',credentials=credentials))

channel = connection.channel()

channel.exchange_declare(exchange='topic_logs',
                         exchange_type='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]...
" % 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()

# python topic_recvive.py #             使用" # "号收所有
# python topic_recvive.py mysql.*       使用"mysql.* "号收来自mysql的信息
# python topic_recvive.py mysql.error.*       使用"mysql.error.* "号收来自mysql的错误信息
# python topic_recvive.py *.django.*       使用"*.django.* "号收来自所有Django的信息
topic_recv.py

执行代码示例

发送端
python topic_send.py python.error test    发送了一条python的错误信息,错误内容为test
python topic_send.py mysql.info hello     发送了一条mysql的信息,信息内容为hello

接收端
python topic_recvive.py #             使用" # "号收所有
python topic_recvive.py mysql.*       使用"mysql.* "号收来自mysql的信息
python topic_recvive.py mysql.error.*       使用"mysql.error.* "号收来自mysql的错误信息
python topic_recvive.py *.django.*       使用"*.django.* "号收来自所有Django的信息

 四、模拟交给服务端程序执行任务。客户端发送消息交给服务端处理

1)简单模拟服务端任务,处理斐波那契数字。该代码存放于服务端(server)

def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)

2)服务端代码示例

import pika
import time

credentials = pika.PlainCredentials('rabbitadmin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.111.230',credentials=credentials))



channel = connection.channel()

channel.queue_declare(queue='rpc_queue_test')


def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)


def on_request(ch, method, props, body):
    n = int(body)

    print(" [.] fib(%s)" % n)
    response = fib(n)

    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id= 
                                                         props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)     # 一次处理一个任务
channel.basic_consume(on_request, queue='rpc_queue_test')

print(" [x] Awaiting RPC requests")
channel.start_consuming()
rpc_server.py

3)客户端代码示例

import pika
import uuid


class FibonacciRpcClient(object):
    def __init__(self):


        credentials = pika.PlainCredentials('rabbitadmin', '123456')
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(
            '192.168.111.230', credentials=credentials))
        channel = self.connection.channel()
        self.channel = self.connection.channel()

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

        self.channel.basic_consume(self.on_response, no_ack=True,#准备接受命令结果
                                   queue=self.callback_queue)


    def on_response(self, ch, method, props, body):
        """"callback方法"""
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4()) #唯一标识符
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue_test',
                                   properties=pika.BasicProperties(
                                       reply_to=self.callback_queue,
                                       correlation_id=self.corr_id,
                                   ),
                                   body=str(n))

        count  = 0
        while self.response is None:
            self.connection.process_data_events() #检查队列里有没有新消息,但不会阻塞
            count +=1
            print("check...",count)

        return int(self.response)


fibonacci_rpc = FibonacciRpcClient()
#1 1 2 3 5 8 13 21  34
#1 2 3 4 5 6 7  8   9
print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(6)
print(" [.] Got %r" % response)
rpc_client.py

原文链接:https://www.cnblogs.com/alex3714/articles/5248247.html

原文地址:https://www.cnblogs.com/linu/p/10228357.html