python对RabbitMQ的简单入门使用教程

apt-get update
apt-get install erlang
apt-get install rabbitmq-server


#启动rabbitmq： service rabbitmq-server start
#停止rabbitmq： service rabbitmq-server stop
#重启rabbitmq： service rabbitmq-server restart

#启动rabbitmq插件：rabbitmq-plugins enable rabbitmq_management

#添加新用户
rabbitmqctl add_user 用户名 密码

#给指定用户添加管理员权限
rabbitmqctl set_user_tags 用户名 administrator

给用户添加权限
rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"

pip install pika
#皮卡皮卡，哈哈

# coding=utf-8
### 生产者

import pika
import time

user_info = pika.PlainCredentials('root', 'root')#用户名和密码
connection = pika.BlockinGConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))#连接服务器上的RabbitMQ服务

# 创建一个channel
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，官方推荐，每次使用时都可以加上这句
channel.queue_declare(queue='hello')

for i in range(0, 100):
    channel.basic_publish(exchange='',#当前是一个简单模式，所以这里设置为空字符串就可以了
                          routing_key='hello',# 指定消息要发送到哪个queue
                          body='{}'.fORMat(i)# 指定要发送的消息
                          )
    time.sleep(1)
    
# 关闭连接
# connection.close()

# coding=utf-8
### 消费者

import pika

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，生产者和消费者都做这一步的好处是
# 这样生产者和消费者就没有必要的先后启动顺序了
channel.queue_declare(queue='hello')


# 回调函数
def callback(ch, method, properties, body):
    print('消费者收到:{}'.format(body))

# channel: 包含channel的一切属性和方法
# method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
# properties: basic_publish 通过 properties 传入的参数
# body: basic_publish发送的消息


channel.basic_consume(queue='hello',  # 接收指定queue的消息
                      auto_ack=True,  # 指定为True，表示消息接收到后自动给消息发送方回复确认，已收到消息
                      on_message_callback=callback  # 设置收到消息的回调函数
                      )

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

# 一直处于等待接收消息的状态，如果没收到消息就一直处于阻塞状态，收到消息就调用上面的回调函数
channel.start_consuming()

# coding=utf-8
### 生产者
import pika
import time

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))

# 创建一个channel
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，官方推荐，每次使用时都可以加上这句
channel.queue_declare(queue='durable_queue',durable=True)
#PS：这里不同种队列不允许名字相同

for i in range(0, 100):
    channel.basic_publish(exchange='',
                          routing_key='durable_queue',
                          body='{}'.format(i),
                          properties=pika.BasicProperties(delivery_mode=2)
                          )
# 关闭连接
# connection.close()

# coding=utf-8
### 消费者

import pika
import time

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，生产者和消费者都做这一步的好处是
# 这样生产者和消费者就没有必要的先后启动顺序了
channel.queue_declare(queue='queue')


# 回调函数
def callback(ch, method, properties, body):
    time.sleep(5)
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print('消费者收到:{}'.format(body.decode('utf-8')))


# channel: 包含channel的一切属性和方法
# method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
# properties: basic_publish 通过 properties 传入的参数
# body: basic_publish发送的消息


channel.basic_consume(queue='queue',  # 接收指定queue的消息
                      auto_ack=False,  # 指定为False，表示取消自动应答，交由回调函数手动应答
                      on_message_callback=callback  # 设置收到消息的回调函数
                      )

# 应答的本质是告诉消息队列可以将这条消息销毁了

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

# 一直处于等待接收消息的状态，如果没收到消息就一直处于阻塞状态，收到消息就调用上面的回调函数
channel.start_consuming()

# coding=utf-8
### 消费者

import pika
import time

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，生产者和消费者都做这一步的好处是
# 这样生产者和消费者就没有必要的先后启动顺序了
channel.queue_declare(queue='queue')


# 回调函数
def callback(ch, method, properties, body):
    time.sleep(0)#通过设置休眠时间来模拟不同消费者的处理时间
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print('消费者收到:{}'.format(body.decode('utf-8')))


# prefetch_count表示接收的消息数量，当我接收的消息没有处理完（用basic_ack标记消息已处理完毕）之前不会再接收新的消息了
channel.basic_qos(prefetch_count=1)  # 还有就是这个设置必须在basic_consume之上，否则不生效

channel.basic_consume(queue='queue',  # 接收指定queue的消息
                      auto_ack=False,  # 指定为False，表示取消自动应答，交由回调函数手动应答
                      on_message_callback=callback  # 设置收到消息的回调函数
                      )

# 应答的本质是告诉消息队列可以将这条消息销毁了

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

# 一直处于等待接收消息的状态，如果没收到消息就一直处于阻塞状态，收到消息就调用上面的回调函数
channel.start_consuming()

#生产者
import pika

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 创建一个指定名称的交换机，并指定类型为fanout，用于将接收到的消息广播到所有queue中
channel.exchange_declare(exchange='交换机', exchange_type='fanout')

# 将消息发送给指定的交换机，在fanout类型中，routing_key=''表示不用发送到指定queue中，
# 而是将发送到绑定到此交换机的所有queue
channel.basic_publish(exchange='交换机', routing_key='', body='这是一条测试消息')

#消费者
import pika

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

channel.exchange_declare(exchange='交换机', exchange_type='fanout')

# 使用RabbitMQ给自己生成一个专有的queue
result = channel.queue_declare(queue='333')
# result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
# 这里如果设置exclusive=True参数，那么该队列就是一个只有队列，在消费者结束后，该专有队列也会自动清除，如果queue=''没有设置名字的话，那么就会自动生成一个
# 不会重复的队列名

# 将queue绑定到指定交换机
channel.queue_bind(exchange='交换机', queue=queue_name)

print(' [*] Waiting for  message.')


def callback(ch, method, properties, body):
    print("消费者收到:{}".format(body.decode('utf-8')))

channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()

import pika
import uuid
import JSON

class RPC(object):

    def __init__(self):
        self.call_id = None
        self.response = None
        user_info = pika.PlainCredentials('root', 'root')
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
        self.channel = self.connection.channel()

        # 创建一个此客户端专用的queue，用于接收服务端发过来的消息
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue

        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):
        # 判断接收到的response是否属于对应request
        if self.call_id == props.correlation_id:
            self.response = json.loads(body.decode('utf-8')).get('result')

    def call(self, func, param):
        self.response = None
        self.call_id = str(uuid.uuid4())  # 为该消息指定uuid，类似于请求id
        self.channel.queue_declare(queue='rpc_queue')
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',  # 将消息发送到该queue
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,  # 从该queue中取消息
                correlation_id=self.call_id,  # 为此次消息指定uuid
            ),
            body=json.dumps(
                {
                    'func': func,
                    'param': {'a': param[0], 'b': param[1]}
                }
            )
        )
        
        self.connection.process_data_events(time_limit=3)# 与start_consuming()相似，可以设置超时参数
        return self.response

rpc = RPC()

print("发送消息到消费者,等待返回结果")

response = rpc.call(func='del', param=(1, 2))

print("收到来自消费者返回的结果:{}".format(response))

import pika
import json

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))

channel = connection.channel()

# 指定接收消息的queue
channel.queue_declare(queue='rpc_queue')

def add_number(a, b):
    return a + b

def del_num(a, b):
    return a - b

execute_map = {
    'add': add_number,
    'del': del_num
}

def on_request(ch, method, props, body):
    body = json.loads(body.decode('utf-8'))
    func = body.get('func')
    param = body.get('param')
    result = execute_map.get(func)(param.get('a'), param.get('b'))
    print('进行{}运算，并将结果返回个消费者'.format(func))

    ch.basic_publish(exchange='',  # 使用默认交换机
                     routing_key=props.reply_to,  # response发送到该queue
                     properties=pika.BasicProperties(
                         correlation_id=props.correlation_id),  # 使用correlation_id让此response与请求消息对应起来
                     body=json.dumps({'result': result}))

    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
# 从rpc_queue中取消息，然后使用on_request进行处理
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

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