用twisted创建tcp socket长链接amf server服务
?
#Email/gtalk:zealzpc@gmail.com
这是pyamf安装包里的例子，总的来说用pyamf来写server的各种例子还是很全的，包括django，gae，twisted，web2py等等，基本python常用的网络框架都有。现在很多网页游戏都是前端flash，那么和server端通信基本就用 amf了，如果server端的功能只是存储一些统计，或者对数据库的增删改查，那么一般的http短连接服务就能搞定了，以上提到的这些框架也都写起来比较简单，比如可以看django的例子http://www.pyamf.org/tutorials/gateways/django.html，寥寥数行就搞定了，python嘛。
?
?
??但有时候需求不是这样的需要flash和server保持一个长连接来不断的进行通信改怎么搞呢？此时在pyamf的网站上看到了sockey这一段，正是我们想要的。
?
??先看下server端的代码：server.py
?
?
# Copyright (c) The PyAMF Project.# See LICENSE.txt for details."""Example socket server using Twisted.@see: U{Documentation for this example<http://pyamf.org/tutorials/actionscript/socket.html>}@since: 0.1"""try: import twistedexcept ImportError: print "This examples requires the Twisted framework. Download it from http://twistedmatrix.com" raise SystemExitfrom twisted.internet.protocol import Protocol, Factoryfrom twisted.internet import reactorfrom datetime import datetimeimport pyamfclass TimerProtocol(Protocol): interval = 1.0 # 客户端链接到server后，server往客户端发送时间的间隔 encoding = pyamf.AMF3 timeout = 20 #客户端链接到server后多少时间不操作就断开链接的timeout def __init__(self): self.started = False #设置编码器 self.encoder = pyamf.get_encoder(self.encoding)、 #设置server端将数据编码成amf后存放的缓存地址 self.stream = self.encoder.stream def connectionLost(self, reason): Protocol.connectionLost(self, reason) print "locst connection:",reason #客户端没断开一个链接，总连接数-1 self.factory.number_of_connections -= 1 print "number_of_connections:",self.factory.number_of_connections def connectionMade(self): #如果服务器连接数超过最大连接数，拒绝新链接建立 if self.factory.number_of_connections >= self.factory.max_connections: self.transport.write('Too many connections, try again later') self.transport.loseConnection() return #总连接数+1 self.factory.number_of_connections += 1 self.timeout_deferred = reactor.callLater(TimerProtocol.timeout, self.transport.loseConnection) def dataReceived(self, data): #去除server收到client数据两端的空格 data = data.strip() #如果收到的是'start'命令 if data == 'start': # start sending a date object that contains the current time if not self.started: self.start() elif data == 'stop': self.stop() #每次执行完客户端请求后重置timeout，重新开始计算无操作时间。 if self.timeout_deferred: self.timeout_deferred.cancel() self.timeout_deferred = reactor.callLater(TimerProtocol.timeout, self.transport.loseConnection) def start(self): self.started = True self.sendTime() def stop(self): self.started = False def sendTime(self): if self.started: #往缓存流里写入信息，用编码器进行amf编码 self.encoder.writeElement(datetime.now()) #返回给客户端编码后的信息 self.transport.write(self.stream.getvalue()) #重置缓存流 self.stream.truncate() #每隔self.interval的时间再发送一次amf信息 reactor.callLater(self.interval, self.sendTime)class TimerFactory(Factory): protocol = TimerProtocol #最大链接数 max_connections = 1000 def __init__(self): self.number_of_connections = 0class SocketPolicyProtocol(Protocol): """ Serves strict policy file for Flash Player >= 9,0,124. @see: U{http://adobe.com/go/strict_policy_files} """ def connectionMade(self): self.buffer = '' def dataReceived(self, data): self.buffer += data if self.buffer.startswith('<policy-file-request/>'): self.transport.write(self.factory.getPolicyFile(self)) self.transport.loseConnection()class SocketPolicyFactory(Factory): protocol = SocketPolicyProtocol def __init__(self, policy_file): """ @param policy_file: Path to the policy file definition """ self.policy_file = policy_file def getPolicyFile(self, protocol): return open(self.policy_file, 'rt').read()#设置域名，端口。host = 'localhost'appPort = 8000policyPort = 843policyFile = 'socket-policy.xml'if __name__ == '__main__': from optparse import OptionParser #设置server启动选项 parser = OptionParser() parser.add_option("--host", default=host, dest="host", help="host address [default: %default]") parser.add_option("-a", "--app-port", default=appPort, dest="app_port", help="Application port number [default: %default]") parser.add_option("-p", "--policy-port", default=policyPort, dest="policy_port", help="Socket policy port number [default: %default]") parser.add_option("-f", "--policy-file", default=policyFile, dest="policy_file", help="Location of socket policy file [default: %default]") (opt, args) = parser.parse_args() print "Running Socket AMF gateway on %s:%s" % (opt.host, opt.app_port) print "Running Policy file server on %s:%s" % (opt.host, opt.policy_port) reactor.listenTCP(int(opt.app_port), TimerFactory(), interface=opt.host) reactor.listenTCP(int(opt.policy_port), SocketPolicyFactory(opt.policy_file), interface=opt.host) reactor.run()?
?
?
里面最主要干活的就是TimerProtocol这个类。
?
在例子中还提供了另外一个文件：time.tac。懂twisted的人应该就知道了，这是一份twistd脚本，可以直接通过twisted自带的twistd来执行。里面import了 ?server.py中已经写好的TimerFactory，然后定义了一个application类，这样就可以通过twistd命令来执行了，twistd是什么可以参考官方文档http://twistedmatrix.com/documents/current/core/howto/application.html?本文就不细说了。
?
debug模式启动server：twistd -noy timer.tac
?
daemo模式启动server：twistd -oy timer.tac
?
你也可以直接启动server.py文件：python server.py、
?
区别就是第二种直接后台运行了在linux，你看不到报错和调试信息，当然windows上不行。
?
?
?
# Copyright (c) The PyAMF Project.# See LICENSE for details.import sys, ossys.path.append(os.getcwd())from twisted.application import internet, servicefrom server import TimerFactory, SocketPolicyFactoryfrom server import appPort, policyPorttimer = TimerFactory()policy = SocketPolicyFactory('socket-policy.xml')# this is the important bitapplication = service.Application('pyamf-socket-example')timerService = internet.TCPServer(appPort, timer)socketPolicyService = internet.TCPServer(policyPort, policy)timerService.setServiceParent(application)socketPolicyService.setServiceParent(application)??
最后就是找个客户端来连接测测，例子里也提供了一个写好的flash，你打开通一个目录下的index.html就可以看看效果了。
?
?
?
??例子里也给你准备了python的客户端。
?
# Copyright (c) The PyAMF Project.# See LICENSE for details."""Python client for socket example.@since: 0.5"""import socketimport pyamffrom server import appPort, hostclass AmfSocketClient(object): def __init__(self): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) def connect(self, host, port): print "Connecting to socket server on %s:%d" % (host, port) try: self.sock.connect((host, port)) print "Connected to server.\n" except socket.error, e: raise Exception("Can't connect: %s" % e[1]) def start(self): msg = '' # tell server we started listening print "send request: start" try: self.sock.send('start') except socket.error, e: raise Exception("Can't connect: %s" % e[1]) while len(msg) < 1024: # read from server amf = self.sock.recv(1024) if amf == '': print "Connection closed." msg = msg + amf for obj in pyamf.decode(amf): print obj return msg def stop(self): print "send request: stop" self.sock.send('stop')if __name__ == '__main__': from optparse import OptionParser parser = OptionParser() parser.add_option("-p", "--port", default=appPort, dest="port", help="port number [default: %default]") parser.add_option("--host", default=host, dest="host", help="host address [default: %default]") (options, args) = parser.parse_args() host = options.host port = int(options.port) client = AmfSocketClient() client.connect(host, port) try: client.start() except KeyboardInterrupt: client.stop() ?原文地址：http://www.pyamf.org/tutorials/actionscript/socket.html
代码就在pyamf安装包的pyamf?/?doc?/?tutorials?/?examples?/?actionscript?/ socket这个目录里。
?
?
欢迎讨论。
1 楼 simomo 2011-07-01 我用的是amfast,这个支持长轮训和streamingchannel不过，貌似pyamf现在也是用C写编码器