Python下的twisted框架入门指引
什么是twisted？
twisted是一个用python语言写的事件驱动的网络框架，他支持很多种协议，包括UDP,TCP,TLS和其他应用层协议，比如HTTP，SMTP，NNTM，IRC，XMPP/Jabber。非常好的一点是twisted实现和很多应用层的协议，开发人员可以直接只用这些协议的实现。其实要修改Twisted的SSH服务器端实现非常简单。很多时候，开发人员需要实现protocol类。
一个Twisted程序由reactor发起的主循环和一些回调函数组成。当事件发生了，比如一个client连接到了server，这时候服务器端的事件会被触发执行。
用Twisted写一个简单的TCP服务器
下面的代码是一个TCPServer，这个server记录客户端发来的数据信息。
====code1.py====
importsys
fromtwisted.internet.protocolimportServerFactory
fromtwisted.protocols.basicimportLineReceiver
fromtwisted.pythonimportlog
fromtwisted.internetimportreactor
classCmdProtocol(LineReceiver):
delimiter='\n'
defconnectionMade(self):
self.client_ip=self.transport.getPeer()[1]
log.msg("Clientconnectionfrom%s"%self.client_ip)
iflen(self.factory.clients)>=self.factory.clients_max:
log.msg("Toomanyconnections.bye!")
self.client_ip=None
self.transport.loseConnection()
else:
self.factory.clients.append(self.client_ip)
defconnectionLost(self,reason):
log.msg('Lostclientconnection.Reason:%s'%reason)
ifself.client_ip:
self.factory.clients.remove(self.client_ip)
deflineReceived(self,line):
log.msg('Cmdreceivedfrom%s:%s'%(self.client_ip,line))
classMyFactory(ServerFactory):
protocol=CmdProtocol
def__init__(self,clients_max=10):
self.clients_max=clients_max
self.clients=[]
log.startLogging(sys.stdout)
reactor.listenTCP(9999,MyFactory(2))
reactor.run()
下面的代码至关重要：
fromtwisted.internetimportreactor reactor.run()
这两行代码会启动reator的主循环。
在上面的代码中我们创建了"ServerFactory"类，这个工厂类负责返回“CmdProtocol”的实例。每一个连接都由实例化的“CmdProtocol”实例来做处理。Twisted的reactor会在TCP连接上后自动创建CmdProtocol的实例。如你所见，protocol类的方法都对应着一种事件处理。
当client连上server之后会触发“connectionMade"方法，在这个方法中你可以做一些鉴权之类的操作，也可以限制客户端的连接总数。每一个protocol的实例都有一个工厂的引用，使用self.factory可以访问所在的工厂实例。
上面实现的”CmdProtocol“是twisted.protocols.basic.LineReceiver的子类，LineReceiver类会将客户端发送的数据按照换行符分隔，每到一个换行符都会触发lineReceived方法。稍后我们可以增强LineReceived来解析命令。
Twisted实现了自己的日志系统，这里我们配置将日志输出到stdout
当执行reactor.listenTCP时我们将工厂绑定到了9999端口开始监听。
user@lab:~/TMP$pythoncode1.py 2011-08-2913:32:32+0200[-]Logopened. 2011-08-2913:32:32+0200[-]__main__.MyFactorystartingon9999 2011-08-2913:32:32+0200[-]Startingfactory<__main__.MyFactoryinstanceat0x227e320 2011-08-2913:32:35+0200[__main__.MyFactory]Clientconnectionfrom127.0.0.1 2011-08-2913:32:38+0200[CmdProtocol,0,127.0.0.1]Cmdreceivedfrom127.0.0.1:helloserver
使用Twisted来调用外部进程
下面我们给前面的server添加一个命令，通过这个命令可以读取/var/log/syslog的内容
importsys
importos
fromtwisted.internet.protocolimportServerFactory,ProcessProtocol
fromtwisted.protocols.basicimportLineReceiver
fromtwisted.pythonimportlog
fromtwisted.internetimportreactor
classTailProtocol(ProcessProtocol):
def__init__(self,write_callback):
self.write=write_callback
defoutReceived(self,data):
self.write("Beginlastlog\n")
data=[lineforlineindata.split('\n')ifnotline.startswith('==')]
fordindata:
self.write(d+'\n')
self.write("Endlastlog\n")
defprocessEnded(self,reason):
ifreason.value.exitCode!=0:
log.msg(reason)
classCmdProtocol(LineReceiver):
delimiter='\n'
defprocessCmd(self,line):
ifline.startswith('lastlog'):
tailProtocol=TailProtocol(self.transport.write)
reactor.spawnProcess(tailProtocol,'/usr/bin/tail',args=['/usr/bin/tail','-10','/var/log/syslog'])
elifline.startswith('exit'):
self.transport.loseConnection()
else:
self.transport.write('Commandnotfound.\n')
defconnectionMade(self):
self.client_ip=self.transport.getPeer()[1]
log.msg("Clientconnectionfrom%s"%self.client_ip)
iflen(self.factory.clients)>=self.factory.clients_max:
log.msg("Toomanyconnections.bye!")
self.client_ip=None
self.transport.loseConnection()
else:
self.factory.clients.append(self.client_ip)
defconnectionLost(self,reason):
log.msg('Lostclientconnection.Reason:%s'%reason)
ifself.client_ip:
self.factory.clients.remove(self.client_ip)
deflineReceived(self,line):
log.msg('Cmdreceivedfrom%s:%s'%(self.client_ip,line))
self.processCmd(line)
classMyFactory(ServerFactory):
protocol=CmdProtocol
def__init__(self,clients_max=10):
self.clients_max=clients_max
self.clients=[]
log.startLogging(sys.stdout)
reactor.listenTCP(9999,MyFactory(2))
reactor.run()
在上面的代码中，没从客户端接收到一行内容后会执行processCmd方法，如果收到的一行内容是exit命令，那么服务器端会断开连接，如果收到的是lastlog，我们要吐出一个子进程来执行tail命令，并将tail命令的输出重定向到客户端。这里我们需要实现ProcessProtocol类，需要重写该类的processEnded方法和outReceived方法。在tail命令有输出时会执行outReceived方法，当进程退出时会执行processEnded方法。
如下是执行结果样例：
user@lab:~/TMP$pythoncode2.py 2011-08-2915:13:38+0200[-]Logopened. 2011-08-2915:13:38+0200[-]__main__.MyFactorystartingon9999 2011-08-2915:13:38+0200[-]Startingfactory<__main__.MyFactoryinstanceat0x1a5a3f8> 2011-08-2915:13:47+0200[__main__.MyFactory]Clientconnectionfrom127.0.0.1 2011-08-2915:13:58+0200[CmdProtocol,0,127.0.0.1]Cmdreceivedfrom127.0.0.1:test 2011-08-2915:14:02+0200[CmdProtocol,0,127.0.0.1]Cmdreceivedfrom127.0.0.1:lastlog 2011-08-2915:14:05+0200[CmdProtocol,0,127.0.0.1]Cmdreceivedfrom127.0.0.1:exit 2011-08-2915:14:05+0200[CmdProtocol,0,127.0.0.1]Lostclientconnection.Reason:[Failureinstance:Traceback(failurewithnoframes):<class'twisted.internet.error.ConnectionDone'>:Connectionwasclosedcleanly.
可以使用下面的命令作为客户端发起命令：
user@lab:~$netcat127.0.0.19999 test Commandnotfound. lastlog Beginlastlog Aug2915:02:03labsSMTP[5919]:Unabletolocatemail Aug2915:02:03labsSMTP[5919]:Cannotopenmail:25 Aug2915:02:03labCRON[4945]:(CRON)error(grandchild#4947failedwithexitstatus1) Aug2915:02:03labsSMTP[5922]:Unabletolocatemail Aug2915:02:03labsSMTP[5922]:Cannotopenmail:25 Aug2915:02:03labCRON[4945]:(logcheck)MAIL(mailed1byteofoutput;butgotstatus0x0001,#012) Aug2915:05:01labCRON[5925]:(root)CMD(command-vdebian-sa1>/dev/null&&debian-sa111) Aug2915:10:01labCRON[5930]:(root)CMD(test-x/usr/lib/atsar/atsa1&&/usr/lib/atsar/atsa1) Aug2915:10:01labCRON[5928]:(CRON)error(grandchild#5930failedwithexitstatus1) Aug2915:13:21labpulseaudio[3361]:ratelimit.c:387eventssuppressed Endlastlog exit
使用Deferred对象
reactor是一个循环，这个循环在等待事件的发生。这里的事件可以是数据库操作，也可以是长时间的计算操作。只要这些操作可以返回一个Deferred对象。Deferred对象可以自动得在事件发生时触发回调函数。reactor会block当前代码的执行。
现在我们要使用Defferred对象来计算SHA1哈希。
importsys
importos
importhashlib
fromtwisted.internet.protocolimportServerFactory,ProcessProtocol
fromtwisted.protocols.basicimportLineReceiver
fromtwisted.pythonimportlog
fromtwisted.internetimportreactor,threads
classTailProtocol(ProcessProtocol):
def__init__(self,write_callback):
self.write=write_callback
defoutReceived(self,data):
self.write("Beginlastlog\n")
data=[lineforlineindata.split('\n')ifnotline.startswith('==')]
fordindata:
self.write(d+'\n')
self.write("Endlastlog\n")
defprocessEnded(self,reason):
ifreason.value.exitCode!=0:
log.msg(reason)
classHashCompute(object):
def__init__(self,path,write_callback):
self.path=path
self.write=write_callback
defblockingMethod(self):
os.path.isfile(self.path)
data=file(self.path).read()
#uncommenttoaddmoredelay
#importtime
#time.sleep(10)
returnhashlib.sha1(data).hexdigest()
defcompute(self):
d=threads.deferToThread(self.blockingMethod)
d.addCallback(self.ret)
d.addErrback(self.err)
defret(self,hdata):
self.write("Filehashis:%s\n"%hdata)
deferr(self,failure):
self.write("Anerroroccured:%s\n"%failure.getErrorMessage())
classCmdProtocol(LineReceiver):
delimiter='\n'
defprocessCmd(self,line):
ifline.startswith('lastlog'):
tailProtocol=TailProtocol(self.transport.write)
reactor.spawnProcess(tailProtocol,'/usr/bin/tail',args=['/usr/bin/tail','-10','/var/log/syslog'])
elifline.startswith('comphash'):
try:
useless,path=line.split('')
except:
self.transport.write('Pleaseprovideapath.\n')
return
hc=HashCompute(path,self.transport.write)
hc.compute()
elifline.startswith('exit'):
self.transport.loseConnection()
else:
self.transport.write('Commandnotfound.\n')
defconnectionMade(self):
self.client_ip=self.transport.getPeer()[1]
log.msg("Clientconnectionfrom%s"%self.client_ip)
iflen(self.factory.clients)>=self.factory.clients_max:
log.msg("Toomanyconnections.bye!")
self.client_ip=None
self.transport.loseConnection()
else:
self.factory.clients.append(self.client_ip)
defconnectionLost(self,reason):
log.msg('Lostclientconnection.Reason:%s'%reason)
ifself.client_ip:
self.factory.clients.remove(self.client_ip)
deflineReceived(self,line):
log.msg('Cmdreceivedfrom%s:%s'%(self.client_ip,line))
self.processCmd(line)
classMyFactory(ServerFactory):
protocol=CmdProtocol
def__init__(self,clients_max=10):
self.clients_max=clients_max
self.clients=[]
log.startLogging(sys.stdout)
reactor.listenTCP(9999,MyFactory(2))
reactor.run()
blockingMethod从文件系统读取一个文件计算SHA1，这里我们使用twisted的deferToThread方法，这个方法返回一个Deferred对象。这里的Deferred对象是调用后马上就返回了，这样主进程就可以继续执行处理其他的事件。当传给deferToThread的方法执行完毕后会马上触发其回调函数。如果执行中出错，blockingMethod方法会抛出异常。如果成功执行会通过hdata的ret返回计算的结果。
推荐的twisted阅读资料
http://twistedmatrix.com/documents/current/core/howto/defer.htmlhttp://twistedmatrix.com/documents/current/core/howto/process.htmlhttp://twistedmatrix.com/documents/current/core/howto/servers.html
API文档：
http://twistedmatrix.com/documents/current/api/twisted.html