pip install pika

# 生產(chǎn)者代碼
import pika


credentials = pika.PlainCredentials('chuan', '123') # mq用戶名和密碼，沒有則需要自己創(chuàng)建
# 虛擬隊列需要指定參數(shù) virtual_host，如果是默認(rèn)的可以不填。
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',
                                port=5672,
                                virtual_host='/',
                                credentials=credentials))

# 建立rabbit協(xié)議的通道
channel = connection.channel()
# 聲明消息隊列，消息將在這個隊列傳遞，如不存在，則創(chuàng)建。durable指定隊列是否持久化
channel.queue_declare(queue='python-test', durable=False)

# message不能直接發(fā)送給queue，需經(jīng)exchange到達(dá)queue，此處使用以空字符串標(biāo)識的默認(rèn)的exchange
# 向隊列插入數(shù)值 routing_key是隊列名
channel.basic_publish(exchange='',
           routing_key='python-test',
           body='Hello world！2')
# 關(guān)閉與rabbitmq server的連接
connection.close()

# 消費者代碼
import pika

credentials = pika.PlainCredentials('chuan', '123')
# BlockingConnection:同步模式
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',
                                port=5672,
                                virtual_host='/',
                              credentials=credentials))
channel = connection.channel()
# 申明消息隊列。當(dāng)不確定生產(chǎn)者和消費者哪個先啟動時，可以兩邊重復(fù)聲明消息隊列。
channel.queue_declare(queue='python-test', durable=False)
# 定義一個回調(diào)函數(shù)來處理消息隊列中的消息，這里是打印出來
def callback(ch, method, properties, body):
  # 手動發(fā)送確認(rèn)消息
  ch.basic_ack(delivery_tag=method.delivery_tag)
  print(body.decode())
  # 告訴生產(chǎn)者，消費者已收到消息

# 告訴rabbitmq，用callback來接收消息
# 默認(rèn)情況下是要對消息進(jìn)行確認(rèn)的，以防止消息丟失。
# 此處將auto_ack明確指明為True，不對消息進(jìn)行確認(rèn)。
channel.basic_consume('python-test',
           on_message_callback=callback)
           # auto_ack=True) # 自動發(fā)送確認(rèn)消息
# 開始接收信息，并進(jìn)入阻塞狀態(tài)，隊列里有信息才會調(diào)用callback進(jìn)行處理
channel.start_consuming()

# 生產(chǎn)者代碼
import pika


credentials = pika.PlainCredentials('chuan', '123') # mq用戶名和密碼
# 虛擬隊列需要指定參數(shù) virtual_host，如果是默認(rèn)的可以不填。
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',
                                port=5672,
                                virtual_host='/',
                                credentials=credentials))

# 建立rabbit協(xié)議的通道
channel = connection.channel()
# 聲明消息隊列，消息將在這個隊列傳遞，如不存在，則創(chuàng)建。durable指定隊列是否持久化。確保沒有確認(rèn)的消息不會丟失
channel.queue_declare(queue='rabbitmqtest', durable=True)

# message不能直接發(fā)送給queue，需經(jīng)exchange到達(dá)queue，此處使用以空字符串標(biāo)識的默認(rèn)的exchange
# 向隊列插入數(shù)值 routing_key是隊列名
# basic_publish的properties參數(shù)指定message的屬性。此處delivery_mode=2指明message為持久的
for i in range(10):
  channel.basic_publish(exchange='',
             routing_key='python-test',
             body='Hello world！%s' % i,
             properties=pika.BasicProperties(delivery_mode=2))
# 關(guān)閉與rabbitmq server的連接
connection.close()

# 消費者代碼,consume1與consume2
import pika
import time

credentials = pika.PlainCredentials('chuan', '123')
# BlockingConnection:同步模式
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',
                                port=5672,
                                virtual_host='/',
                                credentials=credentials))
channel = connection.channel()
# 申明消息隊列。當(dāng)不確定生產(chǎn)者和消費者哪個先啟動時，可以兩邊重復(fù)聲明消息隊列。
channel.queue_declare(queue='rabbitmqtest', durable=True)
# 定義一個回調(diào)函數(shù)來處理消息隊列中的消息，這里是打印出來
def callback(ch, method, properties, body):
  # 手動發(fā)送確認(rèn)消息
  time.sleep(10)
  print(body.decode())
  # 告訴生產(chǎn)者，消費者已收到消息
  ch.basic_ack(delivery_tag=method.delivery_tag)

# 如果該消費者的channel上未確認(rèn)的消息數(shù)達(dá)到了prefetch_count數(shù)，則不向該消費者發(fā)送消息
channel.basic_qos(prefetch_count=1)
# 告訴rabbitmq，用callback來接收消息
# 默認(rèn)情況下是要對消息進(jìn)行確認(rèn)的，以防止消息丟失。
# 此處將no_ack明確指明為True，不對消息進(jìn)行確認(rèn)。
channel.basic_consume('python-test',
           on_message_callback=callback)
           # auto_ack=True) # 自動發(fā)送確認(rèn)消息
# 開始接收信息，并進(jìn)入阻塞狀態(tài)，隊列里有信息才會調(diào)用callback進(jìn)行處理
channel.start_consuming()

# 生產(chǎn)者代碼
import pika


credentials = pika.PlainCredentials('chuan', '123') # mq用戶名和密碼
# 虛擬隊列需要指定參數(shù) virtual_host，如果是默認(rèn)的可以不填。
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',
                                port=5672,
                                virtual_host='/',
                                credentials=credentials))
# 建立rabbit協(xié)議的通道
channel = connection.channel()
# fanout: 所有綁定到此exchange的queue都可以接收消息（實時廣播）
# direct: 通過routingKey和exchange決定的那一組的queue可以接收消息（有選擇接受）
# topic： 所有符合routingKey(此時可以是一個表達(dá)式)的routingKey所bind的queue可以接收消息（更細(xì)致的過濾）
channel.exchange_declare('logs', exchange_type='fanout')


#因為是fanout廣播類型的exchange，這里無需指定routing_key
for i in range(10):
  channel.basic_publish(exchange='logs',
             routing_key='',
             body='Hello world！%s' % i)

# 關(guān)閉與rabbitmq server的連接
connection.close()

import pika

credentials = pika.PlainCredentials('chuan', '123')
# BlockingConnection:同步模式
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',
                                port=5672,
                                virtual_host='/',
                                credentials=credentials))
channel = connection.channel()

#作為好的習(xí)慣，在producer和consumer中分別聲明一次以保證所要使用的exchange存在
channel.exchange_declare(exchange='logs',
             exchange_type='fanout')

# 隨機(jī)生成一個新的空的queue，將exclusive置為True，這樣在consumer從RabbitMQ斷開后會刪除該queue
# 是排他的。
result = channel.queue_declare('', exclusive=True)

# 用于獲取臨時queue的name
queue_name = result.method.queue

# exchange與queue之間的關(guān)系成為binding
# binding告訴exchange將message發(fā)送該哪些queue
channel.queue_bind(exchange='logs',
          queue=queue_name)

# 定義一個回調(diào)函數(shù)來處理消息隊列中的消息，這里是打印出來
def callback(ch, method, properties, body):
  # 手動發(fā)送確認(rèn)消息
  print(body.decode())
  # 告訴生產(chǎn)者，消費者已收到消息
  #ch.basic_ack(delivery_tag=method.delivery_tag)

# 如果該消費者的channel上未確認(rèn)的消息數(shù)達(dá)到了prefetch_count數(shù)，則不向該消費者發(fā)送消息
channel.basic_qos(prefetch_count=1)
# 告訴rabbitmq，用callback來接收消息
# 默認(rèn)情況下是要對消息進(jìn)行確認(rèn)的，以防止消息丟失。
# 此處將no_ack明確指明為True，不對消息進(jìn)行確認(rèn)。
channel.basic_consume(queue=queue_name,
           on_message_callback=callback,
           auto_ack=True) # 自動發(fā)送確認(rèn)消息
# 開始接收信息，并進(jìn)入阻塞狀態(tài)，隊列里有信息才會調(diào)用callback進(jìn)行處理
channel.start_consuming()

# 生產(chǎn)者代碼，測試命令可以使用：python produce.py error 404error
import pika
import sys

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

# 聲明一個名為direct_logs的direct類型的exchange
# direct類型的exchange
channel.exchange_declare(exchange='direct_logs',
             exchange_type='direct')

# 從命令行獲取basic_publish的配置參數(shù)
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'

# 向名為direct_logs的exchage按照設(shè)置的routing_key發(fā)送message
channel.basic_publish(exchange='direct_logs',
           routing_key=severity,
           body=message)

print(" [x] Sent %r:%r" % (severity, message))
connection.close()

# 消費者代碼，測試可以使用：python consume.py error
import pika
import sys

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

# 聲明一個名為direct_logs類型為direct的exchange
# 同時在producer和consumer中聲明exchage或queue是個好習(xí)慣，以保證其存在
channel.exchange_declare(exchange='direct_logs',
             exchange_type='direct')

result = channel.queue_declare('', exclusive=True)
queue_name = result.method.queue

# 從命令行獲取參數(shù):routing_key
severities = sys.argv[1:]
if not severities:
  print(sys.stderr, "Usage: %s [info] [warning] [error]" % (sys.argv[0],))
  sys.exit(1)

for severity in severities:
  # exchange和queue之間的binding可接受routing_key參數(shù)
  # fanout類型的exchange直接忽略該參數(shù)。direct類型的exchange精確匹配該關(guān)鍵字進(jìn)行message路由
  # 一個消費者可以綁定多個routing_key
  # Exchange就是根據(jù)這個RoutingKey和當(dāng)前Exchange所有綁定的BindingKey做匹配，
  # 如果滿足要求，就往BindingKey所綁定的Queue發(fā)送消息
  channel.queue_bind(exchange='direct_logs',
            queue=queue_name,
            routing_key=severity)

def callback(ch, method, properties, body):
  print(" [x] %r:%r" % (method.routing_key, body,))


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

channel.start_consuming()

# 生產(chǎn)者代碼，基本不變，只需將exchange_type改為topic（測試：python produce.py rabbitmq.red 
# red color is my favorite
import pika
import sys

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

# 聲明一個名為direct_logs的direct類型的exchange
# direct類型的exchange
channel.exchange_declare(exchange='topic_logs',
             exchange_type='topic')

# 從命令行獲取basic_publish的配置參數(shù)
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'

# 向名為direct_logs的exchange按照設(shè)置的routing_key發(fā)送message
channel.basic_publish(exchange='topic_logs',
           routing_key=severity,
           body=message)

print(" [x] Sent %r:%r" % (severity, message))
connection.close()

# 消費者代碼，（測試：python consume.py *.red）
import pika
import sys

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

# 聲明一個名為direct_logs類型為direct的exchange
# 同時在producer和consumer中聲明exchage或queue是個好習(xí)慣，以保證其存在
channel.exchange_declare(exchange='topic_logs',
             exchange_type='topic')

result = channel.queue_declare('', exclusive=True)
queue_name = result.method.queue

# 從命令行獲取參數(shù):routing_key
severities = sys.argv[1:]
if not severities:
  print(sys.stderr, "Usage: %s [info] [warning] [error]" % (sys.argv[0],))
  sys.exit(1)

for severity in severities:
  # exchange和queue之間的binding可接受routing_key參數(shù)
  # fanout類型的exchange直接忽略該參數(shù)。direct類型的exchange精確匹配該關(guān)鍵字進(jìn)行message路由
  # 一個消費者可以綁定多個routing_key
  # Exchange就是根據(jù)這個RoutingKey和當(dāng)前Exchange所有綁定的BindingKey做匹配，
  # 如果滿足要求，就往BindingKey所綁定的Queue發(fā)送消息
  channel.queue_bind(exchange='topic_logs',
            queue=queue_name,
            routing_key=severity)

def callback(ch, method, properties, body):
  print(" [x] %r:%r" % (method.routing_key, body,))


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

channel.start_consuming()

# 生產(chǎn)者代碼
import pika
import uuid


# 在一個類中封裝了connection建立、queue聲明、consumer配置、回調(diào)函數(shù)等
class FibonacciRpcClient(object):
  def __init__(self):
    # 建立到RabbitMQ Server的connection
    self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

    self.channel = self.connection.channel()

    # 聲明一個臨時的回調(diào)隊列
    result = self.channel.queue_declare('', exclusive=True)
    self._queue = result.method.queue

    # 此處client既是producer又是consumer，因此要配置consume參數(shù)
    # 這里的指明從client自己創(chuàng)建的臨時隊列中接收消息
    # 并使用on_response函數(shù)處理消息
    # 不對消息進(jìn)行確認(rèn)
    self.channel.basic_consume(queue=self._queue,
                  on_message_callback=self.on_response,
                  auto_ack=True)
    self.response = None
    self.corr_id = None

  # 定義回調(diào)函數(shù)
  # 比較類的corr_id屬性與props中corr_id屬性的值
  # 若相同則response屬性為接收到的message
  def on_response(self, ch, method, props, body):
    if self.corr_id == props.correlation_id:
      self.response = body

  def call(self, n):
    # 初始化response和corr_id屬性
    self.corr_id = str(uuid.uuid4())

    # 使用默認(rèn)exchange向server中定義的rpc_queue發(fā)送消息
    # 在properties中指定replay_to屬性和correlation_id屬性用于告知遠(yuǎn)程server
    # correlation_id屬性用于匹配request和response
    self.channel.basic_publish(exchange='',
                  routing_key='rpc_queue',
                  properties=pika.BasicProperties(
                    reply_to=self._queue,
                    correlation_id=self.corr_id,
                  ),
                  # message需為字符串
                  body=str(n))

    while self.response is None:
      self.connection.process_data_events()

    return int(self.response)


# 生成類的實例
fibonacci_rpc = FibonacciRpcClient()

print(" [x] Requesting fib(30)")
# 調(diào)用實例的call方法
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)

# 消費者代碼，這里以生成斐波那契數(shù)列為例
import pika

# 建立到達(dá)RabbitMQ Server的connection
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 聲明一個名為rpc_queue的queue
channel.queue_declare(queue='rpc_queue')

# 計算指定數(shù)字的斐波那契數(shù)
def fib(n):
  if n == 0:
    return 0
  elif n == 1:
    return 1
  else:
    return fib(n - 1) + fib(n - 2)

# 回調(diào)函數(shù)，從queue接收到message后調(diào)用該函數(shù)進(jìn)行處理
def on_request(ch, method, props, body):
  # 由message獲取要計算斐波那契數(shù)的數(shù)字
  n = int(body)
  print(" [.] fib(%s)" % n)
  # 調(diào)用fib函數(shù)獲得計算結(jié)果
  response = fib(n)

  # exchage為空字符串則將message發(fā)送個到routing_key指定的queue
  # 這里queue為回調(diào)函數(shù)參數(shù)props中reply_ro指定的queue
  # 要發(fā)送的message為計算所得的斐波那契數(shù)
  # properties中correlation_id指定為回調(diào)函數(shù)參數(shù)props中co的rrelation_id
  # 最后對消息進(jìn)行確認(rèn)
  ch.basic_publish(exchange='',
           routing_key=props.reply_to,
           properties=pika.BasicProperties(correlation_id=props.correlation_id),
           body=str(response))
  ch.basic_ack(delivery_tag=method.delivery_tag)


# 只有consumer已經(jīng)處理并確認(rèn)了上一條message時queue才分派新的message給它
channel.basic_qos(prefetch_count=1)

# 設(shè)置consumeer參數(shù)，即從哪個queue獲取消息使用哪個函數(shù)進(jìn)行處理，是否對消息進(jìn)行確認(rèn)
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

print(" [x] Awaiting RPC requests")

# 開始接收并處理消息
channel.start_consuming()