Python下的twisted框架入門指引
什么是twisted？
twisted是一個(gè)用python語(yǔ)言寫的事件驅(qū)動(dòng)的網(wǎng)絡(luò)框架，他支持很多種協(xié)議，包括UDP,TCP,TLS和其他應(yīng)用層協(xié)議，比如HTTP，SMTP，NNTM，IRC，XMPP/Jabber。 非常好的一點(diǎn)是twisted實(shí)現(xiàn)和很多應(yīng)用層的協(xié)議，開發(fā)人員可以直接只用這些協(xié)議的實(shí)現(xiàn)。其實(shí)要修改Twisted的SSH服務(wù)器端實(shí)現(xiàn)非常簡(jiǎn)單。很多時(shí)候，開發(fā)人員需要實(shí)現(xiàn)protocol類。
一個(gè)Twisted程序由reactor發(fā)起的主循環(huán)和一些回調(diào)函數(shù)組成。當(dāng)事件發(fā)生了，比如一個(gè)client連接到了server，這時(shí)候服務(wù)器端的事件會(huì)被觸發(fā)執(zhí)行。
用Twisted寫一個(gè)簡(jiǎn)單的TCP服務(wù)器
下面的代碼是一個(gè)TCPServer，這個(gè)server記錄客戶端發(fā)來(lái)的數(shù)據(jù)信息。
==== 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()
下面的代碼至關(guān)重要：
from twisted.internet import reactor reactor.run()
這兩行代碼會(huì)啟動(dòng)reator的主循環(huán)。
在上面的代碼中我們創(chuàng)建了"ServerFactory"類，這個(gè)工廠類負(fù)責(zé)返回“CmdProtocol”的實(shí)例。 每一個(gè)連接都由實(shí)例化的“CmdProtocol”實(shí)例來(lái)做處理。 Twisted的reactor會(huì)在TCP連接上后自動(dòng)創(chuàng)建CmdProtocol的實(shí)例。如你所見，protocol類的方法都對(duì)應(yīng)著一種事件處理。
當(dāng)client連上server之后會(huì)觸發(fā)“connectionMade"方法，在這個(gè)方法中你可以做一些鑒權(quán)之類的操作，也可以限制客戶端的連接總數(shù)。每一個(gè)protocol的實(shí)例都有一個(gè)工廠的引用，使用self.factory可以訪問(wèn)所在的工廠實(shí)例。
上面實(shí)現(xiàn)的”CmdProtocol“是twisted.protocols.basic.LineReceiver的子類，LineReceiver類會(huì)將客戶端發(fā)送的數(shù)據(jù)按照換行符分隔，每到一個(gè)換行符都會(huì)觸發(fā)lineReceived方法。稍后我們可以增強(qiáng)LineReceived來(lái)解析命令。
Twisted實(shí)現(xiàn)了自己的日志系統(tǒng)，這里我們配置將日志輸出到stdout
當(dāng)執(zhí)行reactor.listenTCP時(shí)我們將工廠綁定到了9999端口開始監(jiān)聽。
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來(lái)調(diào)用外部進(jìn)程
下面我們給前面的server添加一個(gè)命令，通過(guò)這個(gè)命令可以讀取/var/log/syslog的內(nèi)容
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()
在上面的代碼中，沒從客戶端接收到一行內(nèi)容后會(huì)執(zhí)行processCmd方法，如果收到的一行內(nèi)容是exit命令，那么服務(wù)器端會(huì)斷開連接，如果收到的是lastlog，我們要吐出一個(gè)子進(jìn)程來(lái)執(zhí)行tail命令，并將tail命令的輸出重定向到客戶端。這里我們需要實(shí)現(xiàn)ProcessProtocol類，需要重寫該類的processEnded方法和outReceived方法。在tail命令有輸出時(shí)會(huì)執(zhí)行outReceived方法，當(dāng)進(jìn)程退出時(shí)會(huì)執(zhí)行processEnded方法。
如下是執(zhí)行結(jié)果樣例：
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): <class 'twisted.internet.error.ConnectionDone'>: Connection was closed cleanly.
可以使用下面的命令作為客戶端發(fā)起命令：
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對(duì)象
reactor是一個(gè)循環(huán)，這個(gè)循環(huán)在等待事件的發(fā)生。 這里的事件可以是數(shù)據(jù)庫(kù)操作，也可以是長(zhǎng)時(shí)間的計(jì)算操作。 只要這些操作可以返回一個(gè)Deferred對(duì)象。Deferred對(duì)象可以自動(dòng)得在事件發(fā)生時(shí)觸發(fā)回調(diào)函數(shù)。reactor會(huì)block當(dāng)前代碼的執(zhí)行。
現(xiàn)在我們要使用Defferred對(duì)象來(lái)計(jì)算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從文件系統(tǒng)讀取一個(gè)文件計(jì)算SHA1，這里我們使用twisted的deferToThread方法，這個(gè)方法返回一個(gè)Deferred對(duì)象。這里的Deferred對(duì)象是調(diào)用后馬上就返回了，這樣主進(jìn)程就可以繼續(xù)執(zhí)行處理其他的事件。當(dāng)傳給deferToThread的方法執(zhí)行完畢后會(huì)馬上觸發(fā)其回調(diào)函數(shù)。如果執(zhí)行中出錯(cuò)，blockingMethod方法會(huì)拋出異常。如果成功執(zhí)行會(huì)通過(guò)hdata的ret返回計(jì)算的結(jié)果。
推薦的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
- 詳解Python的爬蟲框架 Scrapy
- Python flask框架實(shí)現(xiàn)查詢數(shù)據(jù)庫(kù)并顯示數(shù)據(jù)
- Python flask框架實(shí)現(xiàn)瀏覽器點(diǎn)擊自定義跳轉(zhuǎn)頁(yè)面
- Python flask框架如何顯示圖像到web頁(yè)面
- Python的Django框架實(shí)現(xiàn)數(shù)據(jù)庫(kù)查詢(不返回QuerySet的方法)
- Python ORM框架Peewee用法詳解
- 用Python的pandas框架操作Excel文件中的數(shù)據(jù)教程
- Python爬蟲框架Scrapy安裝使用步驟
- 零基礎(chǔ)寫python爬蟲之使用Scrapy框架編寫爬蟲
- 使用Python的Flask框架實(shí)現(xiàn)視頻的流媒體傳輸
- Python單元測(cè)試框架unittest使用方法講解
- 在Linux上安裝Python的Flask框架和創(chuàng)建第一個(gè)app實(shí)例的教程
- 哪種Python框架適合你？簡(jiǎn)單介紹幾種主流Python框架
相關(guān)文章
Pytest生成測(cè)試報(bào)告的實(shí)現(xiàn)
本文介紹了如何使用 pytest-html 插件生成測(cè)試報(bào)告,并提供了詳細(xì)的操作步驟、配置項(xiàng)和示例代碼,具有一定的參考價(jià)值,感興趣的可以了解一下2023-11-11
Python Celery多隊(duì)列配置代碼實(shí)例
這篇文章主要介紹了Python Celery多隊(duì)列配置代碼實(shí)例,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-11-11
python實(shí)現(xiàn)復(fù)制文件到指定目錄
這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)復(fù)制文件到指定的目錄下，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-10-10
Python爬取網(wǎng)易云歌曲評(píng)論實(shí)現(xiàn)詞云圖
這篇文章主要為大家介紹了Python爬取網(wǎng)易云歌曲評(píng)論實(shí)現(xiàn)詞云分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-06-06
Python實(shí)現(xiàn)自動(dòng)化對(duì)Word文檔添加或移除行號(hào)
Word文檔中的行號(hào)（行編號(hào)）功能是對(duì)于精細(xì)化的文檔編輯以及解析非常有用的功能,添加行號(hào)能夠極大地提升文檔的可讀性和定位效率,本文將介紹如何使用Python來(lái)實(shí)現(xiàn)自動(dòng)化對(duì)Word文檔添加或移除行號(hào),為文檔處理工作帶來(lái)便捷,需要的朋友可以參考下2024-07-07
神經(jīng)網(wǎng)絡(luò)(BP)算法Python實(shí)現(xiàn)及應(yīng)用
這篇文章主要為大家詳細(xì)介紹了Python實(shí)現(xiàn)神經(jīng)網(wǎng)絡(luò)(BP)算法及簡(jiǎn)單應(yīng)用，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-04-04
Python深入分析@property裝飾器的應(yīng)用
這篇文章主要介紹了Python @property裝飾器的用法，在Python中，可以通過(guò)@property裝飾器將一個(gè)方法轉(zhuǎn)換為屬性，從而實(shí)現(xiàn)用于計(jì)算的屬性，下面文章圍繞主題展開更多相關(guān)詳情，感興趣的小伙伴可以參考一下2022-07-07
pytorch中backward()方法如何自動(dòng)求梯度
這篇文章主要介紹了pytorch中backward()方法如何自動(dòng)求梯度問(wèn)題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2023-02-02
python 判斷是否為正小數(shù)和正整數(shù)的實(shí)例
這篇文章主要介紹了python 判斷是否為正小數(shù)和正整數(shù)的實(shí)例的相關(guān)資料,這里提供實(shí)例，實(shí)例注釋說(shuō)明很清楚，需要的朋友可以參考下2017-07-07
在python工程中常常需要使用import引入自己編寫的其他模塊,但其它模塊有時(shí)不在同一個(gè)文件夾下,下面這篇文章主要給大家介紹了關(guān)于python導(dǎo)入其他目錄下模塊的四種情況,需要的朋友可以參考下2022-12-12