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方法。

如下是执行结果样例：

?

1

2

3

4

5

6

7

8

9

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.

可以使用下面的命令作为客户端发起命令：

?

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

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哈希。

?

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

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