AmFast源码浅析

一直以来，很好奇amfast是如何实现服务器端基于长/普通轮询通道的主动推送的。最近看源码有了收获，记录如下。

 

 

正本清源——主动推送

 

      正所谓欲将取之必先予之，要想理解服务器端的主动推送，就必须要坚定心中对于http协议：

一次请求，一次回复；没有请求，没有回复！

的信念，坚信除了websocket和flash socket，所有的浏览器上的消息更新都是基于浏览器的主动发出的http请求，所谓的“主动推送”只是个美丽的名字。

 

隐藏的真相——广播即固化

amfast提供的消息广播的接口如下：

 

from amfast.remoting import flex_messages

#使用channel_set对象的publishMessage方法进行广播
# 创建一个待广播的flex消息对象
msg = flex_messages.AsyncMessage(headers=None,
    body="Hello World", destination="topic-name")
#广播刚才创建的flex消息对象
channel_set.publishMessage(msg)


#使用channel_set对象的publishObject方法进行广播
channel_set.publishObject("Hello World", "topic-name",
    sub_topic="sub-topic-name", headers=None, ttl=30000)

 publishMessage和publishObject接口本质上是一样的，publishiObject会将参数包装成一个flex消息，然后利用publishMessage方法发出去。所以以publishMessage方法为例。

我们可以在amfast/remoting/channel.py里发现这个方法的实现：

 

    def publishMessage(self, msg):
        """Publish a pre-formed message.

        arguments:
        ===========
         * msg - AbstractMessage, the Flex message to publish.
        """
        #关键一
        self.subscription_manager.publishMessage(msg)

        if self.notify_connections is True:
            topic = msg.destination
            if hasattr(msg, 'headers') and \
                msg.headers is not None and \
                messaging.AsyncMessage.SUBTOPIC_HEADER in msg.headers:
                sub_topic = msg.headers[messaging.AsyncMessage.SUBTOPIC_HEADER]
            else:
                sub_topic = None
            #关键二
            self.notifyConnections(topic, sub_topic)

 publishMessage接到待广播的消息后，第一步，先将转手将这个msg交给了subscription_manager.publishMessage 处理。我们可以在subscription_manager

找到这个方法的实现：

 

    def publishMessage(self, msg):
        #对消息过期时间做一些处理
        # Update timestamp to current server time.
        # Is this the correct thing to do???
        msg.timestamp = time.time() * 1000
        if msg.timeToLive is None or msg.timeToLive == 0:
            # Set timeToLive if it has not been pre-set.
            msg.timeToLive = self.ttl
        #关键！
        self.persistMessage(msg)

 这是几道手了？？ 待广播的msg在做了一些简单加工后，被传递给了persistMessage方法。

 

这个方法就比较有意思了，它在subscription_manager的基类中并没有实现，而是在subscription_manager基类的三个子类中实现了，按照oop的叫法，这个persistMessage应该算是一个接口了。我们以这个接口在MemcacheSubscriptionManager中的实现为例：

 

    def persistMessage(self, msg):
        """Store a message."""

        topic = msg.destination
        if hasattr(msg, 'headers') and \
            msg.headers is not None and \
            messaging.AsyncMessage.SUBTOPIC_HEADER in msg.headers:
            sub_topic = msg.headers[messaging.AsyncMessage.SUBTOPIC_HEADER]
        else:
            sub_topic = None
        topic = self.getTopicKey(topic, sub_topic)

        # Remove connection data,
        # so that it is not pickled
        tmp_connection = getattr(msg, 'connection', None)
        if tmp_connection is not None:
            msg.connection = None

        key = self.getKeyName(topic, self.MSG_ATTR)
        lock_name = self.getLockName(key)
        self._lock.acquire(lock_name)
        try:
            messages = self.mc.get(key)
            if messages is None:
                messages = []
            messages.append(msg)
            self.mc.set(key, messages)
        finally:
            self._lock.release(lock_name)

            # Restore connection data
            if tmp_connection is not None:
                msg.connection = tmp_connection

 这段代码的作用其实故名思议，就是将这个待广播的msg以topic和subtopic为key存入到memcache缓存系统里。就这么简单。广播来广播去，最后被存起来了。这就是标题：广播即固化的意思。

 

回归本质，原来神马都是浮云

如果清醒的话，就会回想到，我们刚才是在说channel_set对象的publishMessage方法，然后发现msg被n倒手，最终固化到了缓存系统里，下面接着分析channel_set对象的publishMessage方法。

 

    def publishMessage(self, msg):
        """Publish a pre-formed message.

        arguments:
        ===========
         * msg - AbstractMessage, the Flex message to publish.
        """

        self.subscription_manager.publishMessage(msg)

        if self.notify_connections is True:
            topic = msg.destination
            if hasattr(msg, 'headers') and \
                msg.headers is not None and \
                messaging.AsyncMessage.SUBTOPIC_HEADER in msg.headers:
                sub_topic = msg.headers[messaging.AsyncMessage.SUBTOPIC_HEADER]
            else:
                sub_topic = None

            self.notifyConnections(topic, sub_topic)

 下面就是分支判断，如果你在实例化channel_set对象的时候，设置的notify_connection为True时，就会调用notifyConnections方法。

这个publishMessage方法做的事情很简单，就是把msg固化了一下。下面换个思路，就是从request处理的角度分析。

在客户端发来的request的处理过程中，有一个_pollForMessage的方法会被调用（这个方法之前的调用，在这里就不再说了）。这个方法就是关键所在。

 

def _pollForMessage(self, packet, message, connection):
        """Repeatedly polls for a new message until
        message is available or wait_interval is reached.

        This is blocking, and should only be used
        for Channels where each connection is a thread.

        Synchronous servers should override this method.
        """
        # If True, don't store persistent 'last_polled' value every poll operation.
        soft_touch = hasattr(self.channel_set.connection_manager, "softTouchPolled")

        total_time = 0
        poll_secs = float(self.poll_interval) / 1000
        wait_secs = float(self.wait_interval) / 1000
        while True:
            event = threading.Event()
            event.wait(poll_secs)
            msgs = self.channel_set.subscription_manager.pollConnection(connection, soft_touch)
            if len(msgs) > 0:
                if soft_touch is True:
                    # Store 'last_polled' value.
                    connection.touchPolled()
                #这一步，就是传说中的广播了            
                return msgs
            
            total_time += poll_secs
            if total_time > wait_secs or connection.connected is False:
                if soft_touch is True:
                    # Store 'last_polled' value.
                    connection.touchPolled()
                return ()

 它会调用pollConnection方法，这个方法的作用和刚才的persistMsg是相反的，这个方法会根据topic和subtopic为key，从缓存中取出属于这个客户端的msg，然后把这些msg们返回。如此，那些待广播的消息终于被“广播”了~

 

最后可以发现，其实广播也好，主动推送也好，其实是一个异步过程。

一个只负责将待广播的msg固化到数据库，缓存或内存中

另一头只负责按照自己的topic，subtopic去数据库，缓存或内存中将属于自己的msg取走。

 

另外，我们甚至可以根据amfast的格式，给具体某个客户端存些msg，这样就能实现针对某个客户端而不是某些客户端的消息推送了