什么是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 ====
import sys
from twisted.internet.protocol import ServerFactory
from twisted.protocols.basic import LineReceiver
from twisted.python import log
from twisted.internet import reactor
class CmdProtocol(LineReceiver):
delimiter = '\n'
def connectionMade(self):
self.client_ip = self.transport.getPeer()[1]
log.msg("Client connection from %s" % self.client_ip)
if len(self.factory.clients) >= self.factory.clients_max:
log.msg("Too many connections. bye !")
self.client_ip = None
self.transport.loseConnection()
else:
self.factory.clients.append(self.client_ip)
def connectionLost(self, reason):
log.msg('Lost client connection. Reason: %s' % reason)
if self.client_ip:
self.factory.clients.remove(self.client_ip)
def lineReceived(self, line):
log.msg('Cmd received from %s : %s' % (self.client_ip, line))
class MyFactory(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()
下面的代碼至關重要：
from twisted.internet import reactor
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$ python code1.py
2011-08-29 13:32:32+0200 [-] Log opened.
2011-08-29 13:32:32+0200 [-] __main__.MyFactory starting on 9999
2011-08-29 13:32:32+0200 [-] Starting factory <__main__.MyFactory instance at 0x227e320
2011-08-29 13:32:35+0200 [__main__.MyFactory] Client connection from 127.0.0.1
2011-08-29 13:32:38+0200 [CmdProtocol,0,127.0.0.1] Cmd received from 127.0.0.1 : hello server
使用Twisted來調用外部進程
下面我們給前面的server添加一個命令，通過這個命令可以讀取/var/log/syslog的內容
import sys
import os
from twisted.internet.protocol import ServerFactory, ProcessProtocol
from twisted.protocols.basic import LineReceiver
from twisted.python import log
from twisted.internet import reactor
class TailProtocol(ProcessProtocol):
def __init__(self, write_callback):
self.write = write_callback
def outReceived(self, data):
self.write("Begin lastlog\n")
data = [line for line in data.split('\n') if not line.startswith('==')]
for d in data:
self.write(d + '\n')
self.write("End lastlog\n")
def processEnded(self, reason):
if reason.value.exitCode != 0:
log.msg(reason)
class CmdProtocol(LineReceiver):
delimiter = '\n'
def processCmd(self, line):
if line.startswith('lastlog'):
tailProtocol = TailProtocol(self.transport.write)
reactor.spawnProcess(tailProtocol, '/usr/bin/tail', args=['/usr/bin/tail', '-10', '/var/log/syslog'])
elif line.startswith('exit'):
self.transport.loseConnection()
else:
self.transport.write('Command not found.\n')
def connectionMade(self):
self.client_ip = self.transport.getPeer()[1]
log.msg("Client connection from %s" % self.client_ip)
if len(self.factory.clients) >= self.factory.clients_max:
log.msg("Too many connections. bye !")
self.client_ip = None
self.transport.loseConnection()
else:
self.factory.clients.append(self.client_ip)
def connectionLost(self, reason):
log.msg('Lost client connection. Reason: %s' % reason)
if self.client_ip:
self.factory.clients.remove(self.client_ip)
def lineReceived(self, line):
log.msg('Cmd received from %s : %s' % (self.client_ip, line))
self.processCmd(line)
class MyFactory(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$ python code2.py
2011-08-29 15:13:38+0200 [-] Log opened.
2011-08-29 15:13:38+0200 [-] __main__.MyFactory starting on 9999
2011-08-29 15:13:38+0200 [-] Starting factory <__main__.MyFactory instance at 0x1a5a3f8>
2011-08-29 15:13:47+0200 [__main__.MyFactory] Client connection from 127.0.0.1
2011-08-29 15:13:58+0200 [CmdProtocol,0,127.0.0.1] Cmd received from 127.0.0.1 : test
2011-08-29 15:14:02+0200 [CmdProtocol,0,127.0.0.1] Cmd received from 127.0.0.1 : lastlog
2011-08-29 15:14:05+0200 [CmdProtocol,0,127.0.0.1] Cmd received from 127.0.0.1 : exit
2011-08-29 15:14:05+0200 [CmdProtocol,0,127.0.0.1] Lost client connection. Reason: [Failure instance: Traceback (failure with no frames):
: Connection was closed cleanly.
可以使用下面的命令作為客戶端發起命令：
user@lab:~$ netcat 127.0.0.1 9999
test
Command not found.
lastlog
Begin lastlog
Aug 29 15:02:03 lab sSMTP[5919]: Unable to locate mail
Aug 29 15:02:03 lab sSMTP[5919]: Cannot open mail:25
Aug 29 15:02:03 lab CRON[4945]: (CRON) error (grandchild #4947 failed with exit status 1)
Aug 29 15:02:03 lab sSMTP[5922]: Unable to locate mail
Aug 29 15:02:03 lab sSMTP[5922]: Cannot open mail:25
Aug 29 15:02:03 lab CRON[4945]: (logcheck) MAIL (mailed 1 byte of output; but got status 0x0001, #012)
Aug 29 15:05:01 lab CRON[5925]: (root) CMD (command -v debian-sa1 > /dev/null && debian-sa1 1 1)
Aug 29 15:10:01 lab CRON[5930]: (root) CMD (test -x /usr/lib/atsar/atsa1 && /usr/lib/atsar/atsa1)
Aug 29 15:10:01 lab CRON[5928]: (CRON) error (grandchild #5930 failed with exit status 1)
Aug 29 15:13:21 lab pulseaudio[3361]: ratelimit.c: 387 events suppressed
End lastlog
exit
使用Deferred對象
reactor是一個循環，這個循環在等待事件的發生。 這里的事件可以是數據庫操作，也可以是長時間的計算操作。 只要這些操作可以返回一個Deferred對象。Deferred對象可以自動得在事件發生時觸發回調函數。reactor會block當前代碼的執行。
現在我們要使用Defferred對象來計算SHA1哈希。
import sys
import os
import hashlib
from twisted.internet.protocol import ServerFactory, ProcessProtocol
from twisted.protocols.basic import LineReceiver
from twisted.python import log
from twisted.internet import reactor, threads
class TailProtocol(ProcessProtocol):
def __init__(self, write_callback):
self.write = write_callback
def outReceived(self, data):
self.write("Begin lastlog\n")
data = [line for line in data.split('\n') if not line.startswith('==')]
for d in data:
self.write(d + '\n')
self.write("End lastlog\n")
def processEnded(self, reason):
if reason.value.exitCode != 0:
log.msg(reason)
class HashCompute(object):
def __init__(self, path, write_callback):
self.path = path
self.write = write_callback
def blockingMethod(self):
os.path.isfile(self.path)
data = file(self.path).read()
# uncomment to add more delay
# import time
# time.sleep(10)
return hashlib.sha1(data).hexdigest()
def compute(self):
d = threads.deferToThread(self.blockingMethod)
d.addCallback(self.ret)
d.addErrback(self.err)
def ret(self, hdata):
self.write("File hash is : %s\n" % hdata)
def err(self, failure):
self.write("An error occured : %s\n" % failure.getErrorMessage())
class CmdProtocol(LineReceiver):
delimiter = '\n'
def processCmd(self, line):
if line.startswith('lastlog'):
tailProtocol = TailProtocol(self.transport.write)
reactor.spawnProcess(tailProtocol, '/usr/bin/tail', args=['/usr/bin/tail', '-10', '/var/log/syslog'])
elif line.startswith('comphash'):
try:
useless, path = line.split(' ')
except:
self.transport.write('Please provide a path.\n')
return
hc = HashCompute(path, self.transport.write)
hc.compute()
elif line.startswith('exit'):
self.transport.loseConnection()
else:
self.transport.write('Command not found.\n')
def connectionMade(self):
self.client_ip = self.transport.getPeer()[1]
log.msg("Client connection from %s" % self.client_ip)
if len(self.factory.clients) >= self.factory.clients_max:
log.msg("Too many connections. bye !")
self.client_ip = None
self.transport.loseConnection()
else:
self.factory.clients.append(self.client_ip)
def connectionLost(self, reason):
log.msg('Lost client connection. Reason: %s' % reason)
if self.client_ip:
self.factory.clients.remove(self.client_ip)
def lineReceived(self, line):
log.msg('Cmd received from %s : %s' % (self.client_ip, line))
self.processCmd(line)
class MyFactory(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.html http://twistedmatrix.com/documents/current/core/howto/process.html http://twistedmatrix.com/documents/current/core/howto/servers.html
API文檔：
http://twistedmatrix.com/documents/current/api/twisted.html
更多文章、技術交流、商務合作、聯系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
