pika模块操作rabbitmq-server

环境是centos6.9

rabbitmq依赖erlang语言，首先要安装erlang，然后安装rabbitmq-server

启动rabbitmq-server：service rabbitmq-server start

python3安装pika模块：pip3 install pika

 

pika连接rabbitmq示例：

接收端：

import pika, time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

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

def callback(ch, method, properties, body):
    print("-->正在接收数据...")
    print("[x] Received %r" % body)
    ch.basic_ack(delivery_tag = method.delivery_tag)

#no_ack字段告诉rabbitmq是否需要发送消息接收确认
#channel.basic_consume(callback, queue='hello', no_ack=True)

#rabbitmq如果没有接收到ack就会继续发送此条消息
channel.basic_consume(callback, queue='hello')

print("[* Waiting for message. To exit press CTRL-C]")
channel.start_consuming()

　　发送端：

import pika

#建立管道 
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#信息高速通道 
channel = connection.channel()

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

#通过exchange发送消息到queue
channel.basic_publish(exchange='', routing_key='hello', body='Hello World!',
                     properties = pika.BasicProperties(
                        delivery_mode = 2, #消息持久化,服务器断开也会保存消息
                     )
                     )

print("[x] Send 'Hello World!'")

connection.close()

　　fanout: 所有bind到此exchange的queue都可以接收消息

接收端：

import pika
'''
Procuder->exchange->queue->consumer
'''

#创建管道
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
#创建信息高速
channel = connection.channel()
#声明exchange的类型
channel.exchange_declare(exchange='logs', exchange_type='fanout')
#声明queue,不声明queue名字，rabbitmq会自动生成queue名字，断开后自动删除
result = channel.queue_declare(exclusive=True)
#获取queue名字
queue_name = result.method.queue
#消费者绑定queue
channel.queue_bind(exchange='logs', queue=queue_name)

print(queue_name)
print('[*] Waiting fro logs.To exit press CTRL+C')
#回调函数
def callback(ch, method, properties, body):
    print('-->正在接收数据...')
    print('[x] %r' % body)
#接收句柄
channel.basic_consume(callback,
                      queue = queue_name,
                      no_ack = True
                     )
channel.start_consuming()

　　发送端：

import pika
'''
rabbitmq广播
'''

#建立连接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
#建立信息高速
channel = connection.channel()
#声明exchange为fanout类型
channel.exchange_declare(exchange='logs', exchange_type='fanout')
#创建消息内容
message = 'info:Hello World!'
#发送消息句柄
channel.basic_publish(
    exchange='logs',
    routing_key='',
    body=message
)
#打印发送的消息
print('[x] Send %r' % message)
#关闭连接
connection.close()

　　direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息

接收端：

import pika, sys
'''
exchange的direct模式下的广播
'''
#建立通道
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
#建立信息高速
channel = connection.channel()
#声明exchange类型
channel.exchange_declare(
    exchange='direct_logs',
    exchange_type='direct'
)
#获取queue名字
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]\n' % sys.argv[0])
    sys.exit(1)
#把要接收的所有类型，绑定到exchange的queue里
for severity in severities:
    channel.queue_bind(
        exchange='direct_logs',
        queue=queue_name,
        routing_key=severity
    )

print('[x] 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()

　　发送端：

import pika, sys
'''
exchange为direct的广播模式
可以指定
'''

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

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] send %r:%r' % (severity, message))
connection.close()

　　topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息

发送端：

import pika, sys
'''
exchange的topic模式下更精细的广播过滤
'''
#建立通道
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
#建立信息高速
channel = connection.channel()
#声明exchange类型
channel.exchange_declare(
    exchange='topic_logs',
    exchange_type='topic'
)
#获取queue名字
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

#获取指定接收类型
severities = sys.argv[1:]
#如果没有指定接收类型，报错
if not severities:
    sys.stderr.write('Usage: %s [binding_key]......\n' % sys.argv[0])
    sys.exit(1)
#把要接收的所有类型，绑定到exchange的queue里
for severity in severities:
    channel.queue_bind(
        exchange='topic_logs',
        queue=queue_name,
        routing_key=severity
    )

print('[x] 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()

　　接受端：

import pika, sys
'''
exchange为direct的广播模式
可以指定
'''

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anosy.info'

message = ' '.join(sys.argv[2:]) or 'Hello World!'

channel.basic_publish(
    exchange='topic_logs',
    routing_key=routing_key,
    body=message
)

print('[x] send %r:%r' % (routing_key, message))
connection.close()

　　rpc通过rabbitmq接收并返回消息：

客户端：

import pika, uuid
'''
p端发送消息，并接收返回信息
'''

class RPCClient(object):
    def __init__(self):
        #回收消息
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        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, properties, body):
        if self.corr_id == properties.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',
            properties=pika.BasicProperties(
                #把指定回收的队列发送给服务
                reply_to=self.callback_queue,
                #确认id发送到服务端
                correlation_id=self.corr_id
            ),
            body=str(n)
        )
        while self.response is None:
            #和start_consume的区别是此处不会阻塞
            self.connection.process_data_events()
        return int(self.response)
rpcclient = RPCClient()

print('[x] Requesting fib(10)')
response = rpcclient.call(10)
print('[.] Got %r' % response)

　　服务端：

import pika
'''
rpc服务端
思路如下：
接收客户端消息，
加工数据并打包
发送给客户端
'''
def fibs(n):
    if n == 0:
        return 0
    if n == 1:
        return 1
    return fibs(n-1) + fibs(n-2)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

channel.queue_declare(queue='rpc_queue')


def on_request(ch, method, properties, body):
    n = int(body)
    print('[.] fibs(%s)' % n)
    response = fibs(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)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(
    on_request,
    queue='rpc_queue'
)
print('[x] Awaiting RPC requests...')
channel.start_consuming()