Python之旅

一直用AS3+Red5做一些小游戏，感觉挺方便的，特别在做上下线的功能的时候很方便，Red5都帮我们封装好了，所有需要的方法。基本调用就可以，但是在做大并发访问的时候，Red5就成了一个瓶颈，这个时候用socket是很好的解决办法，可以选择java的socket或是AMFsocket都是可以的，但是真正要提高效率，就要Python和C/C++的iocp来实现，今天做了一个简单的例子，我喜欢使用eclipse IDE，要在eclipse中安装python是很方便的，使用eclipse的软件更新 地址 http://pydev.org/updates就可以下载下来就可以在eclipse中使用了，同时需要下载python 的版本到官网上下载python的版本即可。
python有一个Twisted Matrix框架，Twisted Matrix 附带的文档十分详尽，但却很难掌握。让我们从一个简单的服务器开始，并以之为基础进行构建。在最近一篇 developerWorks技巧文章（请参阅 参考资料以获取链接）中，我演示了一个基于 XML 的“Weblog 服务器”，它向客户机提供了 Web 服务器最新点击数的记录流。XML 方面的问题在这里不很重要，但可以将 SocketServer 及其 ThreadingTCPServer 类作为基线。这个未使用 Twisted Matrix 的服务器包括：
from SocketServer import BaseRequestHandler, ThreadingTCPServer
from time import sleep
import sys, socket
from webloglib import log_fields, hit_tag
class WebLogHandler(BaseRequestHandler):
    def handle(self):
        print"Connected from", self.client_address
        self.request.sendall('<hits>')
        try:
            while True:
                for hit in LOG.readlines():
                    self.request.sendall(hit_tag % log_fields(hit))
                sleep(5)
        except socket.error:
            self.request.close()
        print"Disconnected from", self.client_address
if __name__=='__main__':
    global LOG
    LOG = open('access-log')
    LOG.seek(0, 2)     # Start at end of current access log
    srv = ThreadingTCPServer(('',8888), WebLogHandler)
    srv.serve_forever()
除了创建每个客户机线程的开销之外，这个基于 SocketServer 的服务器一个引人注目的特性在于它对其处理程序内的 time.sleep() 使用阻塞调用。对于 Twisted Matrix 的非阻塞 select() 循环，这样的阻塞是不允许的。

第一个非阻塞方法将任何人为的延迟推给客户机，让客户机明确地请求每批新的 Weblog 记录（它也发送一条消息以表明缺少记录，而不是什么都不发送）。这个使用 Twisted Matrix 的服务器看起来类似：

清单 2. twisted-weblog-1.py


    from twisted.internet import reactor
from twisted.internet.protocol import Protocol, Factory
from webloglib import hit_tag, log_fields
class WebLog(Protocol):
    def connectionMade(self):
        print"Connected from", self.transport.client
        self.transport.write('<hits>')
    def dataReceived(self, data):
        newhits = LOG.readlines()
        ifnot newhits:
            self.transport.write('<none/>')
        for hit in newhits:
            self.transport.write(hit_tag % log_fields(hit))
    def connectionLost(self, reason):
        print"Disconnected from", self.transport.client
factory = Factory()
factory.protocol = WebLog
if __name__=='__main__':
    global LOG
    LOG = open('access-log')
    LOG.seek(0, 2)     # Start at end of current access log
    reactor.listenTCP(8888, factory)
    reactor.run()
    


读者应该参考我先前的一篇技巧文章，以了解客户机应用程序的详细信息。但是必须注意下面的更改。主客户机循环增加了两行：

清单 3. 增强的（阻塞）客户机循环


    while 1:
    xml_data = sock.recv(8192)
    parser.feed(xml_data)
    sleep(5)          # Delay before requesting new records
    sock.send('NEW?') # Send signal to indicate readiness

Twisted 服务器的部件

一个 Twisted Matrix 服务器由几个模块化元素组成。在字节流级别，服务器实现了一项协议，这通常是通过继承 twisted.internet.protocol.Protocol 或继承该类先前专门化的某个子类实现的。例如，假设（ twisted.protocols 中的）子类包括 dns 、 ftp 、 gnutella 、 http 、 nntp 、 shoutcast 以及其他许多协议。协议基本上应该知道如何处理连接的建立和断开，以及如何在连接中接收和发送数据。这些职责与基于 SocketServer 的服务器中的职责没有多大区别，差异在于，前者在为每个元素定义方法的模块化方面略胜一筹。

Twisted Matrix 服务器的下一个级别是工厂。在我们的 twisted-weblog-1.py 示例中，工厂除了存储协议以外其实没做别的事情。不过，在较复杂的服务器中，工厂是执行与协议服务器有关的初始化和终止化操作的好地方。最重要的一点可能是，工厂可以在 应用程序中持久存储（我们很快将看到这一点）。

协议和工厂对服务器运行时所处的网络都一无所知。相反， 反应器（reactor）是实际侦听网络的类（它利用其协议的工厂实例来进行侦听）。反应器基本上只是一个侦听给定端口和网络接口的循环（选择哪个端口和网络接口是通过调用诸如 .listenTCP() 、 .listenSSL() 或 .listenUDP() 之类的方法实现的）。Twisted Matrix 中的基本反应器 SelectReactor 运行在单个线程内，这一点是需要明白的；该服务器会针对新数据检查每一个连接，并将数据传递给相关的协议对象。所产生的结果就是， 确实不允许协议对象阻塞，甚至花费的时间太长以至无法完成（必须适当地进行协议编程）。

增强的服务器

让我们设法增强 Twisted Weblog 服务器，以便它遵循 SocketServer-weblog.py 的模式；无须客户机重复请求即可向客户机提供新记录。这里的问题是，向 WebLog(Protocol) 方法中插入 time.sleep() 调用会导致它阻塞，因此是不允许的。在我们这样做的时候，请注意以前的服务器可能会犯错误，因为它们只向一个客户机提供每批新记录。我们猜测，如果您想允许多个客户机监控一个 Weblog，那么您也会希望它们都接收正在进行的更新。

Twisted Matrix 在不阻塞的情况下延迟操作的方法是使用 .callLater() 方法向反应器添加回调。以此方法添加的回调被添加到提供服务的事件队列中，但只有在指定的延迟之后才会真正地对其进行处理。将这两项更改放在一起，增强的 Weblog 服务器看起来类似：

清单 4. twisted-weblog-1.py


    from twisted.internet import reactor
from twisted.internet.protocol import Protocol, Factory
from webloglib import hit_tag, log_fields
import time
class WebLog(Protocol):
    def connectionMade(self):
        print"Connected from", self.transport.client
        self.transport.write('<hits>')
        self.ts = time.time()
        self.newHits()
    def newHits(self):
        for hit in self.factory.records:
            if self.ts <= hit[0]:
                self.transport.write(hit_tag % log_fields(hit[1]))
        self.ts = time.time()
        reactor.callLater(5, self.newHits)
    def connectionLost(self, reason):
        print"Disconnected from", self.transport.client
class WebLogFactory(Factory):
    protocol = WebLog
    def __init__(self, fname):
        self.fname = fname
        self.records = []
    def startFactory(self):
        self.fp = open(self.fname)
        self.fp.seek(0, 2) # Start at end of current access log
        self.updateRecords()
    def updateRecords(self):
        ts = time.time()
        for rec in self.fp.readlines():
            self.records.append((ts, rec))
        self.records = self.records[-100:]  # Only keep last 100 hits
        reactor.callLater(1, self.updateRecords)
    def stopFactory(self):
        self.fp.close()
if __name__=='__main__':
    reactor.listenTCP(8888, WebLogFactory('access-log'))
    reactor.run()
    


在这个示例中，我们定义了一个定制工厂，并将一些初始化代码从 _main_ 块移到了该工厂。还要注意的是，该服务器的客户机不需要（也不应该）休眠或发送新的请求 — 实际上，我使用的客户机应用程序就是我在 XML 技巧文章中讨论过的客户机应用程序（请参阅 参考资料）。

工厂和协议在各自的定制方法 .updatedRecords() 和 .newHits() 中使用了相同的技术。即，如果方法想要定期运行，那么其最后一行可以调度该方法在指定的延迟以后重复运行。表面上看来，该模式很像递归 — 但它不是递归（而且重复调度不需要一定在最后一行进行，可以在您期望的地方进行调度）。例如，方法 .newHits() 简单地让控制反应器循环知道它希望再过 5 秒钟后被调用，但该方法本身却终止了。我们并不要求方法只能调度自己 — 它可以调度所期望的任何事情，如果愿意的话，也可以将工厂和协议以外的函数添加到反应器循环。


--------------------------------------------------------------------------------
回页首
持久性和调度

除了 reactor.callLater() 调度以外，Twisted Matrix 还包含一个通用类 twisted.internet.defer.Deferred 。实际上， 延迟是对象被调度回调的泛化，但它们也允许使用诸如链接依赖回调和在这些链接中进行错误条件处理之类的技术。 Deferred 对象背后的思想是：当您调用一个方法时，我们不等待其结果（结果可能要过一会儿才出来），该方法可以立即返回一个 Deferred 对象，而反应器／调度程序稍后可以重新调用此对象，那时可望可以得到结果。

我还没有真正地使用 Deferred 对象，但要使它们正常工作好像有些困难。如果您需要等待一个阻塞操作 — 比如，对于来自远程数据库查询的结果 — 您不会确切地知道在可以使用结果之前究竟要等待多长时间。 Deferred 对象 确实有一个超时机制，但我要在今后的文章才讨论这一机制。感兴趣的读者至少应该知道，Twisted Matrix 开发人员已经试图提供一个标准 API 来包装阻塞操作。当然，最坏的情形是回退到使用线程来进行阻塞操作，因为这些操作确实无法转换成异步回调。

Twisted Matrix 服务器另外一个重要元素是它们对持久性提供了方便的支持。反应器是一个监控 I/O 事件并对这些事件做出响应的循环。应用程序类似于增强的反应器，能够将其状态进行 pickle 处理（即序列化），以便用于随后的重新启动。而且，可以将应用程序“有状态地”保存到“.tap”文件，并且可以使用工具 twistd 对其进行管理和监控。这里有一个简单的示例，演示了其用法（它是根据 Twisted 文档的 OneTimeKey 示例进行建模的）。该服务器将不同的 Fibonacci 数传递给所有感兴趣的客户机，而不会在它们之间重复这些数字 — 即使服务器被停止然后被启动：

清单 5. fib_server.py


    from twisted.internet.app import Application
from twisted.internet.protocol import Protocol, Factory
class Fibonacci(Protocol):
    "Serve a sequence of Fibonacci numbers to all requesters"def dataReceived(self, data):
        self.factory.new = self.factory.a + self.factory.b
        self.transport.write('%d' % self.factory.new)
        self.factory.a = self.factory.b
        self.factory.b = self.factory.new
def main():
    import fib_server    # Use script as namespace
    f = Factory()
    f.protocol = fib_server.Fibonacci
    f.a, f.b = 1, 1
    application = Application("Fibonacci")
    application.listenTCP(8888, f)
    application.save()
if'__main__' == __name__:
    main()
    


您可以看到，我们所做的所有工作主要是用 application 替换 reactor 。虽然类 Application 也有一个 .run() 方法，但我们仍然使用其 .save() 方法来创建一个 Fibonacci.tap 文件。运行该服务器的操作如下所示：

清单 6. 运行 fib_server.py


% python fib_server.py
% twistd -f Fibonacci.tap
...let server run, then shut it down...
% kill `cat twistd.pid`
...re-start server where it left off...
% twistd -f Fibonacci-shutdown.tap
...serve numbers where we left off...



连接到该服务器的客户机如果只是间歇地需要新数字，而不需要尽快地得到新数字的话，那么它应该在其循环中使用 time.sleep() 。显然，更有用的服务器可以提供更有趣的有状态数据流。
我的个人网站
http://www.webdesign-cn.com
http://www.levs.cn