上個項目中用到了ActiveMQ，只是簡單應用，安裝完成后直接是用就可以了。由于新項目中一些硬件的限制，需要把消息隊列換成RabbitMQ。

RabbitMQ中的幾種模式和機制比ActiveMQ多多了，根據業務需要，使用RPC實現功能，其中踩過的一些坑，有必要記錄一下了。

上代碼，目錄結構分為 c_server、c_client、c_hanlder:

c_server:

#!/usr/bin/env python# -*- coding:utf-8 -*-import pikaimport timeimport jsonimport ioimport yamls_exchange = input("請輸入交換機名稱->>").decode('utf-8').strip()s_queue = input("輸入消息隊列名稱->>").decode('utf-8').strip()credentials = pika.PlainCredentials('system', 'manager')connection = pika.BlockingConnection(pika.ConnectionParameters(host='XXX.XXX.XXX.XXX',credentials=credentials))# 定義channel = connection.channel()channel.exchange_declare(exchange=s_exchange, exchange_type='direct')channel.queue_declare(queue=s_queue, exclusive=True)channel.queue_bind(queue=s_queue, exchange=s_exchange)def s_manage(content): # 解決unicode轉碼問題 json.JSONDecoder().decode(content) str_content = yaml.safe_load(json.loads(content,encoding='utf-8')) str_res = {  "errorid": 0,  "resp": str_content['cmd'],  "errorcont": "成功" } return json.dumps(str_res)def on_request(ch, method, props, body): response = s_manage(body) ch.basic_publish(exchange='',      routing_key=props.reply_to,      properties=pika.BasicProperties(correlation_id = /               props.correlation_id),      body=response) ch.basic_ack(delivery_tag = method.delivery_tag)channel.basic_qos(prefetch_count=1)channel.basic_consume(on_request, queue=s_queue)print(" [x] Awaiting RPC requests")channel.start_consuming()

c_client:

#!/usr/bin/env python# -*- coding:utf-8 -*-import pikaimport uuidimport jsonimport ioclass RpcClient(object):  def __init__(self):    self.credentials = pika.PlainCredentials('guest', 'guest')    self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='XXX.XXX.XXX.XXX',                                credentials=self.credentials))    self.channel = self.connection.channel()  def on_response(self, ch, method, props, body):    if self.callback_id == props.correlation_id:      self.response = body    ch.basic_ack(delivery_tag=method.delivery_tag)  def get_response(self, callback_queue, callback_id):    '''取隊列里的值，獲取callback_queued的執行結果'''    self.callback_id = callback_id    self.response = None    self.channel.queue_declare('q_manager', durable=True)    self.channel.basic_consume(self.on_response, # 只要收到消息就執行on_response                  queue=callback_queue)    while self.response is None:      self.connection.process_data_events() # 非阻塞版的start_consuming    return self.response  def call(self, queue_name, command, exchange,rout_key): # 命令下發    '''隊列里發送數據'''    # result = self.channel.queue_declare(exclusive=False) #exclusive=False 必須這樣寫    self.callback_queue = 'q_manager' # result.method.queue    self.corr_id = str(uuid.uuid4())    self.channel.basic_publish(exchange=exchange,                  routing_key=queue_name,                  properties=pika.BasicProperties(                    reply_to=self.callback_queue, # 發送返回信息的隊列name                    correlation_id=self.corr_id, # 發送uuid 相當于驗證碼                  ),                  body=command)    return self.callback_queue,self.corr_idclient