Twisted高级话题的学习

Twisted框架之所以高效、强大，是因为其除了提供了基本的通信编程封装，还在设计方法和协议支持上提供了更多地灵活性。这里介绍了非常重要的一部分。

1、延迟调用

2、Twisted使用多线程

3、安全信道

1、延迟调用

延迟（Defer）机制是Twisted框架中用于实现异步编程的体系，使得程序设计可以采用事件驱动的机制。其目的与作用于Tornado的协程类似。

1、基本使用

可以将Twisted中的Defer看作一个管理回调函数的对象，开发者可以向该对象添加需要回调的函数，同时可以指定该组回调函数何时被调用。下面的代码演示了Defer的基本使用方法：

#!/usr/bin/env python
# -*- coding: utf-8 -*-from twisted.internet import defer
from twisted.python import failure
import sysd = defer.Deferred()  # 定义Defer实例# ############以下是Defer回调函数添加阶段############################def printSquare(d):  # 正常处理函数print("Square of %d is %d" % (d, d * d))def processError(f):  # 错误处理函数print("error when process ")d.addCallback(printSquare)  # 添加正常处理回调函数
d.addErrback(processError)  # 添加错误处理回调函数# ##############以下是Defer调用阶段######################################
if len(sys.argv) > 1 and sys.argv[1] == "call_error":f = failure.Failure(Exception("my exception"))d.errback(f)  # 调用错误处理函数processError
else:d.callback(4)  # 调用正常处理函数printSquare(4)

代码围绕着twisted.internet.defer.Deferred对象展开。在Defer中可以管理两种回调函数:正常处理函数和错误处理函数。它们分别由Deferred.addCallback()和Deferred.addErrback()添加到Defer对象中。两种回调处理函数可以分别通过Deferred.callback()和Deferred.errback()进行调用。值得注意的是一个Defer对象在完成添加回调函数的过程后，只能由callback()或errback()进行一次调用。如果试图进行第二次调用，则Teisted框架将会抛出异常。

一个Defer对象可以被赋予多个正常或错误的回调函数，这样形成的函数链将会按顺序调用执行

# !/usr/bin/env python
# -*- coding: utf-8 -*-from twisted.internet import deferd = defer.Deferred()  # 定义Defer实例def printSquare(d):  # 正常处理函数print("Square of %d is %d" % (d, d * d))return ddef processError(f):  # 错误处理函数print("error when process ")def printTwice(d):print("Twice of %d is %d" % (d, 2 * d))return dd.addCallback(printSquare)  # 添加正常处理回调函数
d.addErrback(processError)  # 添加错误处理回调函数
d.addCallback(printTwice)  # 添加第2个正常处理回调函数d.callback(5)

程序输入为：

Square of 5 is 25
Twice of 5 is 10

2、Defer对象详解

1、addCallback(self,callback,*args,**kwargs)

　　回调函数中必须有参数（单个，多个，默认值），不能为空；回调函数被调用时，回调函数第一个参数是defer函数链中前一个正常处理函数的返回结果，其后的参数是在addCallback（）时指定的args和kw参数

2、addErrback(self,errback,*args,**kwargs)

　　和正常回调函数类似，当函数被调用时第一个参数是failure.Failure(Exception('my exception'))的实例，用于说明错误情况

3、addBoth(self,callback,*args,**kwargs)

　　将同一个回调函数同时作为正常处理函数和错误处理函数添加到defer对象中

4、chainDeferred(self,d)

　　Defer对象链接函数用于将另一个defer对象的处理函数和错误处理函数添加到本defer对象中，本函数具有单向性，比如下面的代码：

D1 = defer.Deferred()
D2 = defer.Deferred()D1.chainDeferred(D2

该代码中的D1再被调用时将导致D2对象中的函数链也被调用，而D2对象被调用时将不会导致D1中的函数链被调用。

5、callback(self,result)

　　调用正常处理函数连，result是传递给第一个正常处理回调函数的参数

6、errback(self,fail=None)

　　调用错误处理函数链，fail是传递给第一个错误处理回调函数的参数

7、pause(self)和unpause(self)

　　为Defer对象调用链的暂停和继续pause函数用于暂停一个defer对象中函数链的调用，直到unpause函数被调用后继续

3、defer回调函数链的调用流程

4、结合defer和reactor

将defer对象和reactor的延时调用机制结合在一起，就可以开发出功能强大的异步调用函数,关于这部分的详细学习可以去看Dava的诗歌服务器。http://krondo.com/an-introduction-to-asynchronous-programming-and-twisted/

from twisted.internet import reactor,defer
#####################函数添加阶段def printSquare(d): # 正常处理函数print('square of %d id %d '% (d,d*d))return ddef printTwice(d):print('Twice of %d is %d'%(d,2*d))return ddef makeDefer():d = defer.Deferred()  # 定义defer实例d.addCallback(printSquare) # 添加正常处理回调函数d.addCallback(printTwice) # 添加第二个正常处理回调函数reactor.callLater(2,d.callback,5) # 配置延时2秒调用################defer调用阶段
makeDefer()
reactor.run() #挂起运行

其中的reactor.callLater（）是twisted中关键的异步函数，其函数原型如下：
def reactor.callLater（delay,callable,*args,**kwargs）
第一个参数为延迟时间，之后参数为被调用的函数名及其参数利用callLater（）函数，可以实现定时退出twisted消息循环：

from twisted.internet import reactor,deferreactor.callLater(4,reactor.stop) # 运行4秒后调用reactor.stop
reactor.run()
print('程序退出')

2、Twisted使用多线程

Twisted 提供主线程和辅线程，主线程只有1个，即reactor.run(),辅线程有多个，可以自由配置

Twisted 大多数代码运行在主线程中，dataReceived(),connectionLose()等事件处理函数由主线程调用，如果耗时太长，可以将其移到辅助线程中处理

同时，Twisted框架中大多数内置函数都不是线程安全的，因此需要将内置函数放入主线程中，否则会导致逻辑错误或者系统崩溃。

线程安全：指多个线程在执行同一段代码的时候采用加锁机制，使每次的执行结果和单线程执行的结果都是一样的，不存在执行程序时出现意外结果。
- 线程不安全：是指不提供加锁机制保护，有可能出现多个线程先后更改数据造成所得到的数据是脏数据。

1、使代码运行在主线程中

如果其他线程中需要执行非线程安全的Twisted内置函数，可以使用reactor.callFromThread()函数使代码运行在主线程中：

from twisted.internet import reactor,defer
import Myprotocolprotocol = Myprotocol()
def must_run_in_main_thread(message):protocol.send = Trueprotocol.transport.write(message)def run_in_any_thread():reactor.callFromThread(must_run_in_main_thread,'good')print('the run of must_run_in_main_thread has been finashed')

callFromThread将自己的线程在调用处挂起。直到被调用的函数已经在主线程中完成
注：callFromThread不仅可以是twisted的辅助线程，还可以是twisted主线程，或是python threading库建立的线程

2、在辅助线程中运行

在主线程中遇到比较耗时的处理时，可以用reactor.callInThread()函数建立辅助线程任务：

from twisted.internet import reactor,defer
from twisted.internet.protocol import DatagramProtocoldef long_operation(msg):import time time.sleep(10)print('10秒后获取信息',msg)class Echo(DatagramProtocol):def datagramReceived(self, datagram, addr):# 调用long_operation，在辅助线程中执行，本调用在主线程中立即返回reactor.callInThread(long_operation,datagram.decode('utf8'))protocol = Echo()
reactor.listenUDP(8007,protocol)
reactor.run()

3、配置线程池

可以使用reactor.suggestThreadPoolsize(10)，定义线程数量

3、安全信道

SSL也被成为TLS，为TCP通信加入SSL信道可以认证通信的主体，保证通信内容的私密性和完整性。

SSL信道只能建立在传输层TCP上，无法建立在UDP上。

在进行SSL的通信开发之前，需要安装Python的SSL插件pyOpenSSL:

pip install pyopenssl

从Twisted SSL信道的目的来看，SSL通信可以分为两个级别：

加密信道的SSL通信：是为了保证网络中传输内容的私密性，第三方即使窥探到了全部的通信比特流，也无法破解正式内容
认证客户端身份的SSL通信：除了能加密通道，还能提供客户端的身份认证，即只允许有身份证明的客户端

1、加密信道的SSL通信

安装OpenSSL工具包

sudo apt-get install openssl

sudo apt-get install libssl-dev

OpenSSL命令：/usr/bin/openssl.

配置文件：usr/lib/ssl/*

生成SSL密钥和证书

生成CA证书ca.crt，服务器密钥文件server.key，和服务器证书server.crt

# 生成CA密钥
openssl genrsa -out ca.key 2048# 生成CA证书，本过程还会要求输入证书的所在地，公司等
openssl req -x509 -new -nodes -key ca.key -days 365 -out ca.crt# 生成服务器证书RSA的密钥对
openssl genrsa -out server.key 2048# 生成服务端证书CSR，本过程会要求输入证书所在地，公司等
openssl req -new -key server.key -out server.csr# 生成服务器端证书 ca.crt
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 365

上述命令生成服务器端证书时，必须在common nanme（CN)字段中如实输入站点访问地址，比如定义CN=www.mysite.com,如果通过ip访问就设ip地址。

服务端代码：

from twisted.internet import ssl,reactor
from twisted.internet.protocol import Factory,Protocolclass EchoServer(Protocol):def dataReceived(self, data):self.transport.write(data)if __name__ == '__main__':factory = Factory()factory.protocol = EchoServerreactor.listenSSL(8007,factory,ssl.DefaultOpenSSLContextFactory('../ssl/server.key','../ssl/server.crt')) # 配置密钥文件和证书文件路径reactor.run()

客户端：

from twisted.internet import ssl,reactor
from twisted.internet.protocol import ClientFactoryif __name__ == '__main__':factory = ClientFactory()reactor.connectSSL('192.168.1.10',8000,factory,ssl.ClientContextFactory())# SSL连接reactor.run()

这样TCP服务器和客户端就可以密文通信了

2、认证客户端身份的SSL通信

在应用中除了需要保证通信的私密性服务器还需要防止伪造的客户端与服务器通信，所以需要改造TCP客户端，

客户端身份通过客户端证书进行标识：

from twisted.internet import ssl,reactor
from twisted.internet.protocol import ClientFactoryclass MySSLContext(ssl.ClientContextFactory):def getContext(self):self.method = ssl.SSL.SSLv3_METHOD   #SSL协议版本ctx = ssl.ClientContextFactory.getContext(self)ctx.use_certificate_file('../ssl/client.crt') #客户端证书ctx.use_privatekey_file('../ssl/client.key') # 客户端密钥return ctxif __name__ == '__main__':factory = ClientFactory()reactor.connectSSL('192.168.1.10',8000,factory,MySSLContext())# SSL连接reactor.run()

服务器端需要根据客户端提交的证书验证是否允许其与自己通信：

from twisted.internet import ssl,reactor
from twisted.internet.protocol import Factory,Protocol
from OpenSSL import SSLclass Echo(Protocol):def dataReceived(self, data):self.transport.write(data)def verifyCallback(connection,x509,errnum,errdepth,ok):print("_verify(ok =%d):" % ok)  # CA是否匹配print('subject:',x509.get_subject()) # 客户端证书subjectprint('issuer:',x509.get_issuer()) # 客户端证书发行方print('errnum %s,errdepth %d' % (errnum,errdepth)) # 错误代码return True # 是否允许通信if __name__ == '__main__':factory = Factory()factory.protocol = EchomyContextFactory = ssl.DefaultOpenSSLContextFactory('../ssl/server.key','../ssl/server.crt') # 服务器端的密钥和证书文件ctx = myContextFactory.getContext()ctx.set_verify(ssl.SSL.VERIFY_PEER | ssl.SSL.VERIFY_FAIL_IF_NO_PEER_CERT,verifyCallback) ctx.load_verify_locations('../ssl/ca.crt') # 客户端证书的发行CAreactor.listenSSL(8007,factory,myContextFactory) # 配置密钥文件和证书文件路径reactor.run()