RabbitMQ 是一个基于 高级消息队列协议 (AMQP) 的强大消息中间件。由于 AMQP 规范的通用性，可以轻松地从包括 Python 在内的许多平台连接到它。在本篇博文中，我们将

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

您可以在 http://github.com/gregturn/amqp-demo 找到此博文的所有源代码。这需要您已经根据 您平台的说明 安装并启动了 RabbitMQ。我个人是在我的 Mac OS X (Snow Leopard) 机器上运行的。

顺便说一句

本篇博文中编写的代码仅用于演示目的。请勿依赖这些算法作为财务建议。

闲话不多说，让我们开始写代码吧！

构建股票行情

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

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 中，消息的正文只是一系列字节。存储什么以及如何序列化不属于规范的一部分，而必须在发送方和接收方之间达成一致。在我们的情况下，发布者和订阅者都同意它包含一个 pickled 元组。

创建 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”。这意味着绑定到它的每个队列都将收到消息的副本，而无需在代理端进行昂贵的处理。

您可能会想，为什么正文的 content_type 是“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 对消息的不透明正文进行反序列化，在通道上向代理确认消息的接收，然后启动其关于决定买入还是卖出的算法。
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 代理，声明队列，将其绑定到 fanout 交换机，然后注册回调的基本模式。

但是，让我们不要过于纠结于如何让这个算法在挑选赢家和输家方面做得更好。相反，让我们认识到这使得任何金融公司都可以通过创建唯一的队列，绑定到股票系统的 fanout 交换机，然后编写自己的金融决策算法来轻松订阅股票行情。

用 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 将它们全部组合在一个类中。这就是为什么回调方法与提取我们消息正文的实际方法之间存在解耦。通过提取必要的部分，可以轻松地在这些库之间切换，而无需重写我们的买卖算法。

运行

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

########################################
# 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。核心方法名称相同。几个参数也相同。但更重要的是：架构概念是相同的。选择哪个库现在可以不仅包括技术优点，还包括客户支持、规范遵从性、同步与异步支持以及可用性等。