RabbitMQ 是一个基于 高级消息队列协议 (AMQP) 的功能强大的消息代理。由于 AMQP 规范的特性，它很容易从许多平台连接，包括 Python。在本篇博文中，我们将

• 创建一个简单的股票行情 Python 应用程序
• 创建一个经纪人 Python 应用程序，用于决定何时买卖。
• 比较 pika（RabbitMQ 团队创建的 AMQP 库）和 py-amqplib。

您可以在 http://github.com/gregturn/amqp-demo 找到此博客的所有源代码。这假设您已根据 平台说明 安装了 RabbitMQ 并启动它。就我个人而言，它在我的 Mac OS X 机器（雪豹）上运行。

顺便说一句

本博文中编写的代码仅用于演示目的。不要依赖这些算法来获取财务建议。

言归正传，让我们编写一些代码吧！

构建股票行情

消息解决方案的一个很好的例子是股票行情系统。证券交易所将消息发布到代理，指示股票名称、价格和时间。

import pickle
import random
import time

class Ticker(object):
    def __init__(self, publisher, qname):
        self.publisher = publisher

        # This quickly creates four random stock symbols
        chars = range(ord("A"), ord("Z")+1)
        def random_letter(): return chr(random.choice(chars))
        self.stock_symbols = [random_letter()+random_letter()+random_letter() for i in range(4)]

        self.last_quote = {}
        self.counter = 0
        self.time_format = "%a, %d %b %Y %H:%M:%S +0000"
        self.qname = qname

    def get_quote(self):
        symbol = random.choice(self.stock_symbols)
        if symbol in self.last_quote:
            previous_quote = self.last_quote[symbol]
            new_quote = random.uniform(0.9*previous_quote, 1.1*previous_quote)
            if abs(new_quote) - 0 < 1.0:
                new_quote = 1.0
            self.last_quote[symbol] = new_quote
        else:
            new_quote = random.uniform(10.0, 250.0)
            self.last_quote[symbol] = new_quote
        self.counter += 1
        return (symbol, self.last_quote[symbol], time.gmtime(), self.counter)

    def monitor(self):
        while True:
            quote = self.get_quote()
            print("New quote is %s" % str(quote))
            self.publisher.publish(pickle.dumps((quote[0], quote[1], time.strftime(self.time_format, quote[2]), quote[3])), routing_key="")
            secs = random.uniform(0.1, 0.5)
            #print("Sleeping %s seconds..." % secs)
            time.sleep(secs)

此应用程序随机创建四个股票代码，然后开始创建报价。它最初会在 10.0 到 250.0 之间选择一个随机值，然后继续在先前价格的 90% 到 110% 之间随机调整价格。然后，它会在 0.1 到 0.5 秒之间随机等待，然后再发出下一个报价。此代码设计的一个重要部分是，发布到 AMQP 代理已与股票行情分离。相反，它期望在构建时注入发布者服务。

需要注意的是，我们正在使用 pickle 来序列化我们的股票报价数据元组。在 AMQP 中，消息的主体只是一系列字节。存储的内容以及如何序列化不是规范的一部分，而必须由发送方和接收方约定。在我们的情况下，发布者和订阅者都同意它包含一个 pickle 化的元组。

创建 AMQP 服务

下一步是创建我们的 AMQP 客户端服务。它的目的是使我们能够轻松地隔离与 AMQP 服务器的通信，无论是通过发布还是通过使用事件。

from amqplib import client_0_8 as amqp

class PyAmqpLibPublisher(object):
    def __init__(self, exchange_name):
        self.exchange_name = exchange_name
        self.queue_exists = False

    def publish(self, message, routing_key):
        conn = amqp.Connection(host="127.0.0.1", userid="guest", password="guest", virtual_host="/", insist=False)

        ch = conn.channel()

        ch.exchange_declare(exchange=self.exchange_name, type="fanout", durable=False, auto_delete=False)

        msg = amqp.Message(message)
        msg.properties["content_type"] = "text/plain"
        msg.properties["delivery_mode"] = 2
        ch.basic_publish(exchange=self.exchange_name,
                         routing_key=routing_key,
                         msg=msg)
        ch.close()
        conn.close()

这里要注意的一件重要事情是，声明的交换机类型为“fanout”。这意味着绑定到它的每个队列都会收到消息的副本，而无需在代理端进行昂贵的处理。

您可能想知道为什么主体的内容类型为“text/plain”，因为它是一个序列化消息。这是因为 Python 的 pickle 库以 ASCII 加密的格式编码数据，可以使用任何工具查看而不会导致奇怪的行为。

低买高卖

一些简单明智的建议是在价格低的时候买入，在价格高的时候卖出。在这里，我们将查看一个简单的客户端，它订阅股票报价，收集价格的历史趋势以确定下一个价格是低端还是高端，然后决定买入或卖出。

import pickle
import random
import uuid

class Buyer(object):
    def __init__(self, client, qname, trend=5):
        self.holdings = {}
        self.cash = 100000.0
        self.history = {}
        self.qname = qname
        self.client = client
        self.trend = trend
        self.qname = uuid.uuid4().hex

    def decide_whether_to_buy_or_sell(self, quote):
        symbol, price, date, counter = quote
        #print "Thinking about whether to buy or sell %s at %s" % (symbol, price)

        if symbol not in self.history:
            self.history[symbol] = [price]
        else:
            self.history[symbol].append(price)

        if len(self.history[symbol]) >= self.trend:
            price_low = min(self.history[symbol][-self.trend:])
            price_max = max(self.history[symbol][-self.trend:])
            price_avg = sum(self.history[symbol][-self.trend:])/self.trend
            #print "Recent history of %s is %s" % (symbol, self.history[symbol][-self.trend:])
        else:
            price_low, price_max, price_avg = (-1, -1, -1)
            print "%s quotes until we start deciding whether to buy or sell %s" % (self.trend - len(self.history[symbol]), symbol)
            #print "Recent history of %s is %s" % (symbol, self.history[symbol])

        if price_low == -1: return

        #print "Trending minimum/avg/max of %s is %s-%s-%s" % (symbol, price_low, price_avg, price_max)
        #for symbol in self.holdings.keys():
        #    print "self.history[symbol][-1] = %s" % self.history[symbol][-1]
        #    print "self.holdings[symbol][0] = %s" % self.holdings[symbol][0]
        #    print "Value of %s is %s" % (symbol, float(self.holdings[symbol][0])*self.history[symbol][-1])
        value = sum([self.holdings[symbol][0]*self.history[symbol][-1] for symbol in self.holdings.keys()])
        print "Net worth is %s + %s = %s" % (self.cash, value, self.cash + value)

        if symbol not in self.holdings:
            if price < 1.01*price_low:
                shares_to_buy = random.choice([10, 15, 20, 25, 30])
                print "I don't own any %s yet, and the price is below the trending minimum of %s so I'm buying %s shares." % (symbol, price_low, shares_to_buy)
                cost = shares_to_buy * price
                print "Cost is %s, cash is %s" % (cost, self.cash)
                if cost < self.cash:
                    self.holdings[symbol] = (shares_to_buy, price, cost)
                    self.cash -= cost
                    print "Cash is now %s" % self.cash
                else:
                    print "Unfortunately, I don't have enough cash at this time."
        else:
            if price > self.holdings[symbol][1] and price > 0.99*price_max:
                print "+++++++ Price of %s is higher than my holdings, so I'm going to sell!" % symbol
                sale_value = self.holdings[symbol][0] * price
                print "Sale value is %s" % sale_value
                print "Holdings value is %s" % self.holdings[symbol][2]
                print "Total net is %s" % (sale_value - self.holdings[symbol][2])
                self.cash += sale_value
                print "Cash is now %s" % self.cash
                del self.holdings[symbol]

    def handle_pyamqplib_delivery(self, msg):
        self.handle(msg.delivery_info["channel"], msg.delivery_info["delivery_tag"], msg.body)

    def handle(self, ch, delivery_tag, body):
        quote = pickle.loads(body)
        #print "New price for %s => %s at %s" % quote
        ch.basic_ack(delivery_tag = delivery_tag)
        print "Received message %s" % quote[3]
        self.decide_whether_to_buy_or_sell(quote)

    def monitor(self):
        self.client.monitor(self.qname, self.handle_pyamqplib_delivery)

此客户端将其买卖股票的策略很好地隔离在接收来自 RabbitMQ 的消息的机制之外。

1. monitor 是启动监听新股票报价的主要钩子。它将 handle_pyamqplib_delivery 注册为每次有新报价到达时调用的回调方法。
2. handle_pyamqplib_delivery 提取消息的重要部分并将其传递给 handle。插入此额外方法调用的原因是为了支持用 pika 替换 py-amqplib，我们稍后将介绍。
3. handle 解 pickle 化消息的不透明主体，使用通道向代理确认已收到消息，然后触发其关于决定是否买入或卖出的算法。
4. decide_whether_to_buy_or_sell 拆分股票报价的元组，然后将价格添加到其股票代码历史记录中。它旨在在做出决定之前收集最少的报价。难道不是吗？然后它计算趋势的最小值和最大值，如果价格相对接近最小值，则买入。但是，如果它已经持有股票，则等待价格上涨到其最初购买的价格以上。发生这种情况时，它会卖出。

缺少的部分是 self.client.monitor 函数。self.client 是我们之前编码的 AMQP 服务上的钩子，我们需要一种方法将我们的队列绑定到交换机以接收消息。需要将以下函数添加到 PyAmqpLibPublisher 中。

    def monitor(self, qname, callback):
        conn = amqp.Connection(host="127.0.0.1", userid="guest", password="guest")

        ch = conn.channel()

        if not self.queue_exists:
            ch.queue_declare(queue=qname, durable=False, exclusive=False, auto_delete=False)
            ch.queue_bind(queue=qname, exchange=self.exchange_name)
            print "Binding queue %s to exchange %s" % (qname, self.exchange_name)
            #ch.queue_bind(queue=qname, exchange=self.exchange_name, routing_key=qname)
            self.queue_exists = True

        ch.basic_consume(callback=callback, queue=qname)

        while True:
            ch.wait()
        print 'Close reason:', conn.connection_close

这展示了连接到我们的 RabbitMQ 代理、声明队列、将其绑定到扇出交换机，然后注册回调的基本模式。

但是，让我们不要过分纠结于如何使此算法更好地选择赢家和输家。相反，让我们认识到这使得任何金融公司都可以通过创建一个唯一的队列、绑定到股票系统的扇出交换机，然后编写自己的算法来做出财务决策，从而轻松订阅股票报价。

用 pika 替换 py-amqplib

AMQP 是一个编写良好的规范。它包括 XML 格式，支持自动生成客户端库。这意味着根据规范编码的库很容易互换，并根据其实现的优点进行选择。Python 社区中一个流行的库是 py-amqplib。正如其项目站点上所述，它的一个限制是它会阻塞并且目前不提供并发性。pika 同时提供两者。

重点是，从 py-amqplib 迁移到 pika 实际上非常容易。基于 AMQP 的方法是相同的，底层概念也是相同的。让我们看看如何使用 pika 编写一个备用的 AMQP 服务。

import pika

class PikaPublisher(object):
    def __init__(self, exchange_name):
        self.exchange_name = exchange_name
        self.queue_exists = False

    def publish(self, message, routing_key):
        conn = pika.AsyncoreConnection(pika.ConnectionParameters(
                '127.0.0.1',
                credentials=pika.PlainCredentials('guest', 'guest')))

        ch = conn.channel()

        ch.exchange_declare(exchange=self.exchange_name, type="fanout", durable=False, auto_delete=False)

        ch.basic_publish(exchange=self.exchange_name,
                         routing_key=routing_key,
                         body=message,
                         properties=pika.BasicProperties(
                                content_type = "text/plain",
                                delivery_mode = 2, # persistent
                                ),
                         block_on_flow_control = True)
        ch.close()
        conn.close()

    def monitor(self, qname, callback):
        conn = pika.AsyncoreConnection(pika.ConnectionParameters(
                '127.0.0.1',
                credentials=pika.PlainCredentials('guest', 'guest')))

        ch = conn.channel()

        if not self.queue_exists:
            ch.queue_declare(queue=qname, durable=False, exclusive=False, auto_delete=False)
            ch.queue_bind(queue=qname, exchange=self.exchange_name)
            print "Binding queue %s to exchange %s" % (qname, self.exchange_name)
            #ch.queue_bind(queue=qname, exchange=self.exchange_name, routing_key=qname)
            self.queue_exists = True

        ch.basic_consume(callback, queue=qname)

        pika.asyncore_loop()
        print 'Close reason:', conn.connection_close

这与之前显示的其他服务非常相似。创建连接有点不同，但包含相同的数据部分，例如 broker 的主机以及 username 和 password。basic_publish 略有不同，因为消息及其属性是在方法调用中一起放入的。py-amqplib 在稍微不同的结构中声明整个消息及其属性，然后将其作为单个参数传递给 basic_publish。规范的好处在于知道所有重要部分都在这两个库中。

与 py-amqplib 相比，pika 支持不同的等待机制。py-amqplib 具有阻塞等待，而 pika 同时提供阻塞机制以及使用 Python 的 asyncore 实用程序 进行异步操作的机制。我们可以在以后关于 RabbitMQ 和 Python 的博文中探讨这一点。

这两个库之间回调方法的签名略有不同。我们需要更新我们的经纪人客户端以适当地处理它。

    def handle_pyamqplib_delivery(self, msg):
        self.handle(msg.delivery_info["channel"], msg.delivery_info["delivery_tag"], msg.body)

将其与 pika 的回调方法签名进行比较。

    def handle_pika_delivery(self, ch, method, header, body):
        self.handle(ch, delivery_tag, body)

它们非常接近。重要的部分都在那里。差异基于 pika 拆分消息的部分，而 py-amqplib 将所有内容组合到单个类中。这就是回调方法和实际提取消息主体的 method 之间存在解耦的原因。通过提取必要的部件，可以在这两个库之间切换而无需重写我们的买卖算法。

运行程序

有了所有这些代码，我们需要运行它们。编写一个运行脚本并启动它们很容易。

########################################
# To run this demo using py-amqplib,
# uncomment this block, and  comment out
# the next block.
########################################

#from amqplib_client import *
#publisher = PyAmqpLibPublisher(exchange_name="my_exchange")

########################################
# To run this demo using pika,
# uncomment this block, and comment out
# the previous block
########################################

from pika_client import *
publisher = PikaPublisher(exchange_name="my_exchange")

########################################
# This part doesn't have to change
########################################

from ticker_system import *
ticker = Ticker(publisher, "")
ticker.monitor()

此运行程序可以在运行 py-amqplib 或 pika 版本的股票行情系统之间切换。现在我们只需要一个经纪人服务的运行程序。

########################################
# To run this demo using py-amqplib,
# uncomment this block, and  comment out
# the next block.
########################################

#from amqplib_client import *
#publisher = PyAmqpLibPublisher(exchange_name="my_exchange")

########################################
# To run this demo using pika,
# uncomment this block, and comment out
# the previous block
########################################

from pika_client import *
publisher = PikaPublisher(exchange_name="my_exchange")

########################################
# This part doesn't have to change
########################################

from buy_low_sell_high import *
buyer = Buyer(publisher, "", trend=25)
print "Buyer = %s" % id(buyer)
buyer.monitor()

在以后的博文中，我们可以考虑使用 Pythonic DI 容器运行相同的代码。

良好的规范提供了极佳的选择

AMQP 规范使得可以根据不仅仅是技术优点来选择库。通过将 AMQP 的机制与生成报价以及解析报价的逻辑分开，可以很容易地互换 py-amqplib 和 pika。核心方法名称相同。一些参数相同。但更重要的是：架构概念相同。现在，选择哪个库不仅可以包括技术优点，还可以包括客户支持、规范合规性、同步与异步支持以及可用性等因素。