pip install pika

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


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

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

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

# 消費(fèi)者代碼
import pika

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

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

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


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

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

# message不能直接發(fā)送給queue，需經(jīng)exchange到達(dá)queue，此處使用以空字符串標(biāo)識(shí)的默認(rèn)的exchange
# 向隊(duì)列插入數(shù)值 routing_key是隊(duì)列名
# 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()

# 消費(fèi)者代碼,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()
# 申明消息隊(duì)列。當(dāng)不確定生產(chǎn)者和消費(fèi)者哪個(gè)先啟動(dòng)時(shí)，可以?xún)蛇呏貜?fù)聲明消息隊(duì)列。
channel.queue_declare(queue='rabbitmqtest', durable=True)
# 定義一個(gè)回調(diào)函數(shù)來(lái)處理消息隊(duì)列中的消息，這里是打印出來(lái)
def callback(ch, method, properties, body):
  # 手動(dòng)發(fā)送確認(rèn)消息
  time.sleep(10)
  print(body.decode())
  # 告訴生產(chǎn)者，消費(fèi)者已收到消息
  ch.basic_ack(delivery_tag=method.delivery_tag)

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

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


credentials = pika.PlainCredentials('chuan', '123') # mq用戶(hù)名和密碼
# 虛擬隊(duì)列需要指定參數(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都可以接收消息（實(shí)時(shí)廣播）
# direct: 通過(guò)routingKey和exchange決定的那一組的queue可以接收消息（有選擇接受）
# topic： 所有符合routingKey(此時(shí)可以是一個(gè)表達(dá)式)的routingKey所bind的queue可以接收消息（更細(xì)致的過(guò)濾）
channel.exchange_declare('logs', exchange_type='fanout')


#因?yàn)槭莊anout廣播類(lèi)型的exchange，這里無(wú)需指定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ī)生成一個(gè)新的空的queue，將exclusive置為T(mén)rue，這樣在consumer從RabbitMQ斷開(kāi)后會(huì)刪除該queue
# 是排他的。
result = channel.queue_declare('', exclusive=True)

# 用于獲取臨時(shí)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)

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

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

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

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

# 聲明一個(gè)名為direct_logs的direct類(lèi)型的exchange
# direct類(lèi)型的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()

# 消費(fèi)者代碼，測(cè)試可以使用：python consume.py error
import pika
import sys

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

# 聲明一個(gè)名為direct_logs類(lèi)型為direct的exchange
# 同時(shí)在producer和consumer中聲明exchage或queue是個(gè)好習(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類(lèi)型的exchange直接忽略該參數(shù)。direct類(lèi)型的exchange精確匹配該關(guān)鍵字進(jìn)行message路由
  # 一個(gè)消費(fèi)者可以綁定多個(gè)routing_key
  # Exchange就是根據(jù)這個(gè)RoutingKey和當(dāng)前Exchange所有綁定的BindingKey做匹配，
  # 如果滿(mǎn)足要求，就往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（測(cè)試：python produce.py rabbitmq.red 
# red color is my favorite
import pika
import sys

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

# 聲明一個(gè)名為direct_logs的direct類(lèi)型的exchange
# direct類(lèi)型的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()

# 消費(fèi)者代碼，（測(cè)試：python consume.py *.red）
import pika
import sys

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

# 聲明一個(gè)名為direct_logs類(lèi)型為direct的exchange
# 同時(shí)在producer和consumer中聲明exchage或queue是個(gè)好習(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類(lèi)型的exchange直接忽略該參數(shù)。direct類(lèi)型的exchange精確匹配該關(guān)鍵字進(jìn)行message路由
  # 一個(gè)消費(fèi)者可以綁定多個(gè)routing_key
  # Exchange就是根據(jù)這個(gè)RoutingKey和當(dāng)前Exchange所有綁定的BindingKey做匹配，
  # 如果滿(mǎn)足要求，就往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


# 在一個(gè)類(lèi)中封裝了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()

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

    # 此處client既是producer又是consumer，因此要配置consume參數(shù)
    # 這里的指明從client自己創(chuàng)建的臨時(shí)隊(duì)列中接收消息
    # 并使用on_response函數(shù)處理消息
    # 不對(duì)消息進(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ù)
  # 比較類(lèi)的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)


# 生成類(lèi)的實(shí)例
fibonacci_rpc = FibonacciRpcClient()

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

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

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

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

# 計(jì)算指定數(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獲取要計(jì)算斐波那契數(shù)的數(shù)字
  n = int(body)
  print(" [.] fib(%s)" % n)
  # 調(diào)用fib函數(shù)獲得計(jì)算結(jié)果
  response = fib(n)

  # exchage為空字符串則將message發(fā)送個(gè)到routing_key指定的queue
  # 這里queue為回調(diào)函數(shù)參數(shù)props中reply_ro指定的queue
  # 要發(fā)送的message為計(jì)算所得的斐波那契數(shù)
  # properties中correlation_id指定為回調(diào)函數(shù)參數(shù)props中co的rrelation_id
  # 最后對(duì)消息進(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時(shí)queue才分派新的message給它
channel.basic_qos(prefetch_count=1)

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

print(" [x] Awaiting RPC requests")

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