Part1.1 、RabbitMQ 操作使用

• 先来一个基于Queue实现生产者消费者模型试试水
  

#!/usr/bin/env python3
#coding:utf8

import pika
# ######################### 生产者 #########################
#链接rabbit服务器（localhost是本机，如果是其他服务器请修改为ip地址）
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘))
channel = connection.channel()        #创建频道
channel.queue_declare(queue=‘hello‘)  #创建一个队列名叫hello

channel.basic_publish(exchange=‘‘,       exchange -- 它使我们能够确切地指定消息应该到哪个队列去。 
                  routing_key=‘hello‘,     队列 routing_key是队列名。
                  body=‘Hello World!‘)     body是要插入的内容
print("队列开始")

connection.close() #缓冲区已经flush而且消息已经确认发送到了RabbitMQ中，关闭链接

#!/usr/bin/env python
import pika
# ########################## 消费者 ##########################

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘)) #链接rabbit

channel = connection.channel() #创建频道


提示：如果生产者没有运行创建队列，那么消费者也许就找不到队列了。为了避免这个问题，
所以，消费者也可自行-创建这个队列，，避免报错。
channel.queue_declare(queue=‘hello‘)

#接收消息需要使用callback这个函数来接收，它会被pika库来调用
def callback(ch, method, properties, body):
    print(" [x] 接受信息 %r" % body)

channel.basic_consume(callback,      #scallback是回调函数 如果拿到数据 那么将执行callback函数
                      queue=‘hello‘,     #选择 操作的队列
                      no_ack=True)       #是否为了保证自己的消费端的数据安全，回复一个ack,当MQ没有接收的话，会再次发送。此参数是不回复。
print(‘ [*] 等待信息.退出 Ctrl+c‘)

channel.start_consuming()  #永远循环等待数据处理和callback处理的数据

#coding:utf-8
#!/usr/bin/env python
import pika
# ######################### 生产者 #########################

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘)) #链接rabbit服务器
channel = connection.channel() #创建频道

channel.queue_declare(queue=‘Myqueue‘) #创建一个队列名叫Myqueue

channel.basic_publish(exchange=‘‘,   #向队列插入数值 routing_key是队列名 body是要插入的内容
                   routing_key=‘Myqueue‘,
                  body=‘Hello World!‘)
print("开始队列")
connection.close()

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘)) #链接rabbit
channel = connection.channel() #创建频道

channel.queue_declare(queue=‘Myqueue‘) #如果生产者没有运行创建队列，那么消费者创建队列

def callback(ch, method, properties, body):
    print(" [x] 接收到 %r" % body)
    import time
    time.sleep(10)
    print (‘ok‘)
    ch.basic_ack(delivery_tag = method.delivery_tag) #主要使用此代码

channel.basic_consume(callback,
                      queue=‘Myqueue‘,
                      no_ack=False)

print(‘ [*] 等待接受信息... 退出按 ctrl+c ‘)
channel.start_consuming()

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘)) #链接rabbit服务器
channel = connection.channel() #创建频道

‘‘‘
  提示：两处都要声明下，1、在创建队列时候durable=True  2、在队列使用的参数中 delivery_mode=2 
‘‘‘
channel.queue_declare(queue=‘Myqueue‘, durable=True) 
                                    #创建队列，使用durable方法。
                                       #如果想让队列实现持久化那么加上durable=True 。
                                    
channel.basic_publish(exchange=‘‘,
                      routing_key=‘Myqueue‘,
                      body=‘Hello World!‘,
                      properties=pika.BasicProperties(
                      delivery_mode=2, 
                      #标记我们的消息为持久化的 - 通过设置 delivery_mode 属性为 2
                      #这样必须设置，让消息实现持久化
                      ))

#这个exchange参数就是这个exchange的名字. 空字符串标识默认的或者匿名的exchange：
# 如果存在routing_key, 消息路由到routing_key指定的队列中。
print(" [x] 开始队列‘")
connection.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘))
channel = connection.channel() #创建频道
channel.queue_declare(queue=‘Myqueue‘, durable=True) #创建队列，使用durable方法

def callback(ch, method, properties, body):
    print(" [x] 接受信息 %r" % body)
    import time
    time.sleep(10)
    print (‘ok‘)
    ch.basic_ack(delivery_tag = method.delivery_tag)

    channel.basic_consume(callback,
                    queue=‘Myqueue‘,
                    no_ack=False)

    print(‘ [*] 等待队列. 退出 CTRL+C‘)
    channel.start_consuming()

import pika
import sys  

connection = pika.BlockingConnection(pika.ConnectionParameters(  
    host=‘localhost‘))  
channel = connection.channel()  

channel.queue_declare(queue=‘task_queue‘, durable=True) # 设置队列为持久化的队列
message = ‘ ‘.join(sys.argv[1:]) or "Hello World!"  
channel.basic_publish(exchange=‘‘,  
                     routing_key=‘task_queue‘,  
                     body=message,  
                     properties=pika.BasicProperties(  
                     delivery_mode = 2,  #设置消息为持久化的
                  ))  
print(" [x] Sent %r" % message)  
connection.close()  

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘))
channel = connection.channel()
channel.queue_declare(queue=‘task_queue‘,durable=True)  # 设置队列持久化

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    import time
    time.sleep(10)
    print (‘ok‘)
    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)
                               #消息未处理完前不要发送信息的消息
                                  #表示谁来谁取，不再按照奇偶数排列

channel.basic_consume(callback,
                  queue=‘task_queue‘,
                  no_ack=False)

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