快捷導(dǎo)航

Python下的twisted框架入門(mén)指引

更新時(shí)間：2015年04月15日 17:12:44 投稿：goldensun

這篇文章主要介紹了Python下的twisted框架入門(mén)指引,twisted框架是一個(gè)異步機(jī)制的框架,也被許多Python教程所介紹,需要的朋友可以參考下

什么是twisted？

twisted是一個(gè)用python語(yǔ)言寫(xiě)的事件驅(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é)議，開(kāi)發(fā)人員可以直接只用這些協(xié)議的實(shí)現(xiàn)。其實(shí)要修改Twisted的SSH服務(wù)器端實(shí)現(xiàn)非常簡(jiǎn)單。很多時(shí)候，開(kāi)發(fā)人員需要實(shí)現(xiàn)protocol類(lèi)。

一個(gè)Twisted程序由reactor發(fā)起的主循環(huán)和一些回調(diào)函數(shù)組成。當(dāng)事件發(fā)生了，比如一個(gè)client連接到了server，這時(shí)候服務(wù)器端的事件會(huì)被觸發(fā)執(zhí)行。
用Twisted寫(xiě)一個(gè)簡(jiǎn)單的TCP服務(wù)器

下面的代碼是一個(gè)TCPServer，這個(gè)server記錄客戶(hù)端發(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"類(lèi)，這個(gè)工廠類(lèi)負(fù)責(zé)返回“CmdProtocol”的實(shí)例。每一個(gè)連接都由實(shí)例化的“CmdProtocol”實(shí)例來(lái)做處理。 Twisted的reactor會(huì)在TCP連接上后自動(dòng)創(chuàng)建CmdProtocol的實(shí)例。如你所見(jiàn)，protocol類(lèi)的方法都對(duì)應(yīng)著一種事件處理。

當(dāng)client連上server之后會(huì)觸發(fā)“connectionMade"方法，在這個(gè)方法中你可以做一些鑒權(quán)之類(lèi)的操作，也可以限制客戶(hù)端的連接總數(shù)。每一個(gè)protocol的實(shí)例都有一個(gè)工廠的引用，使用self.factory可以訪問(wèn)所在的工廠實(shí)例。

上面實(shí)現(xiàn)的”CmdProtocol“是twisted.protocols.basic.LineReceiver的子類(lèi)，LineReceiver類(lèi)會(huì)將客戶(hù)端發(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端口開(kāi)始監(jiān)聽(tīng)。

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()

在上面的代碼中，沒(méi)從客戶(hù)端接收到一行內(nèi)容后會(huì)執(zhí)行processCmd方法，如果收到的一行內(nèi)容是exit命令，那么服務(wù)器端會(huì)斷開(kāi)連接，如果收到的是lastlog，我們要吐出一個(gè)子進(jìn)程來(lái)執(zhí)行tail命令，并將tail命令的輸出重定向到客戶(hù)端。這里我們需要實(shí)現(xiàn)ProcessProtocol類(lèi)，需要重寫(xiě)該類(lèi)的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.

可以使用下面的命令作為客戶(hù)端發(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

您可能感興趣的文章: