import pika
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost")) # 建立程序與rabbitmq的連接
channel = connection.channel()
channel.queue_declare(queue='hello') # 定義hello隊(duì)列
channel.basic_publish(exchange='',
      routing_key='hello', # 告訴rabbitmq將消息發(fā)送到hello隊(duì)列中
      body='Hello world!') # 發(fā)送消息的內(nèi)容
print(" [x] Sent 'Hello World!'")
connection.close() # 關(guān)閉與rabbitmq的連接

import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost")) # 建立程序與rabbitmq的連接
channel = connection.channel()
# 在接收端定義隊(duì)列，參數(shù)與發(fā)送端的相同
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
 """
 收到消息調(diào)用callback處理消息
 :param ch:
 :param method:
 :param properties:
 :param body:
 :return:
 """
 print(" [x] received %r" % body)
 # time.sleep(30)
 print("Done....")
channel.basic_consume(callback,
      queue='hello', # 告訴rabbitmq此程序從hello隊(duì)列中接收消息
      no_ack=True)

# channel.basic_consume(callback,
#      queue='hello')
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() # 開始接收，未收到消息阻塞

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)  #使隊(duì)列持久化
message = "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()

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

channel.queue_declare(queue='task_queue', durable=True) #再次申明隊(duì)列，和發(fā)送端參數(shù)應(yīng)一樣
print(' [*] Waiting for messages. To exit press CTRL+C')

def callback(ch, method, properties, body):
 print(" [x] received %r" % body)
 time.sleep(2)
 print(" [x] Done")
 # 因?yàn)闆]有設(shè)置no_ask=True, 所以需要告訴rabbitmq消息已經(jīng)處理完畢，rabbitmq將消息移出隊(duì)列。
 ch.basic_ack(delivery_tag=method.delivery_tag)

#同一時(shí)間worker只接收一條消息，等這條消息處理完在接收下一條
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
      queue='task_queue')


channel.start_consuming()

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 申明一個(gè)exchange,兩個(gè)參數(shù)分別為exchange的名字和類型；當(dāng)exchang='fanout'時(shí)，所有綁定到此exchange的消費(fèi)者隊(duì)列都將收到消息
channel.exchange_declare(exchange='logs',
       exchange_type='fanout')
# 消息可以在命令行啟動(dòng)腳本時(shí)以參數(shù)的形式傳入
# message = ' '.join(sys.argv[1:]) or "info: Hello World!"
message = 'Hello World!'
channel.basic_publish(exchange='logs',
      routing_key='',
      body=message)
print(" [x] Sent %r" % message)
connection.close()

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
       exchange_type='fanout')
# 隨機(jī)生成一個(gè)queue，此queue唯一，且在連接端開后自動(dòng)銷毀
result = channel.queue_declare(exclusive=True)
# 得到隨機(jī)生成消費(fèi)者隊(duì)列的名字
queue_name = result.method.queue
# 將消費(fèi)者隊(duì)列與exchange綁定
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] received %r" % body)

channel.basic_consume(callback,
      queue=queue_name,
      no_ack=True)

channel.start_consuming()

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
       exchange_type='direct')
# 命令行啟動(dòng)時(shí)，以參數(shù)的的形式傳入發(fā)送消息的級(jí)別，未傳怎默認(rèn)設(shè)置未info
# severity = sys.argv[1] if len(sys.argv) > 2 else 'info'
# 命令行啟動(dòng)時(shí)，以參數(shù)的的形式傳入發(fā)送消息的內(nèi)容，未傳怎默認(rèn)設(shè)置Hello World!
# message = ' '.join(sys.argv[2:]) or 'Hello World!'
# 演示使用，實(shí)際運(yùn)用應(yīng)用上面的方式設(shè)置消息級(jí)別
severity = 'info' #作為例子直接將消息的級(jí)別設(shè)置為info
# severity = 'warning'
message = 'Hello World'

#使用exchang的direct模式時(shí)，routing_key的值為消息的級(jí)別
channel.basic_publish(exchange='direct_logs',
      routing_key=severity,
      body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
       exchange_type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
# 命令行啟動(dòng)時(shí)以參數(shù)的形式傳入要接收哪個(gè)級(jí)別的消息，可以傳入多個(gè)級(jí)別
# severities = sys.argv[1:]
# 演示使用，實(shí)際運(yùn)用應(yīng)該用上面的方式指明消息級(jí)別
# 作為演示，直接設(shè)置兩個(gè)接收級(jí)別，info 和 warning
severities = ['info', 'warning']
if not severities:
 """如果要接收消息的級(jí)別不存在則提示用戶輸入級(jí)別并退出程序"""
 sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
 sys.exit(1)
for severity in severities:
 """依次為每個(gè)消息級(jí)別綁定queue"""
 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()

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
       exchange_type='topic')
# 以命令行的方式啟動(dòng)發(fā)送端，以參數(shù)的形式傳入發(fā)送消息類型的關(guān)鍵字
routing_key = sys.argv[1] if len(sys.argv[1]) > 2 else 'anonymous.info' 
# routing_key = 'anonymous.info'
# routing_key = 'abc.orange.abc'
# routing_key = 'abc.abc.rabbit'
# routing_key = 'lazy.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()

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
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:]
# binding_keys = '#'  #接收所有的消息
# binding_keys = ['*.info']  #接收所有以".info"結(jié)尾的消息
# binding_keys = ['*.orange.*'] #接收所有含有".orange."的消息
# binding_keys = ['*.*.rabbit', 'lazy.*'] #接收所有含有兩個(gè)擴(kuò)展名且結(jié)尾是".rabbit"和所有以"lazy."開頭的消息
if not binding_keys:
 sys.stderr.write("Usage: %s [binding_key]...\n" % 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()

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='rpc_queue')
def fib(n):
 """
 計(jì)算斐波那契數(shù)列中第n個(gè)數(shù)的值
 :param n:
 :return:
 """
 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, # 使用客戶端傳來的隊(duì)列向客戶端發(fā)送消息的處理結(jié)果
      properties=pika.BasicProperties(
       correlation_id = props.correlation_id), # 指明處理消息的id 用于客戶端確認(rèn)
      body=str(response))
 ch.basic_ack(delivery_tag = method.delivery_tag) # 未申明no_ack = True, 消息處理完畢需向rabbitmq確認(rèn)
channel.basic_qos(prefetch_count=1) # 每次只處理一條消息
channel.basic_consume(on_request, queue='rpc_queue')
print(" [x] Awaiting RPC requests")
channel.start_consuming() # 開始接收消息，未收到消息處于阻塞狀態(tài)