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

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()