首页 后端开发 Python教程 Python下的twisted框架入门指引

Python下的twisted框架入门指引

Jun 06, 2016 am 11:25 AM
python

什么是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): <class 'twisted.internet.error.ConnectionDone'>: 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

本站声明
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn

热AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智能驱动的应用程序,用于创建逼真的裸体照片

AI Clothes Remover

AI Clothes Remover

用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool

Undress AI Tool

免费脱衣服图片

Clothoff.io

Clothoff.io

AI脱衣机

Video Face Swap

Video Face Swap

使用我们完全免费的人工智能换脸工具轻松在任何视频中换脸!

热工具

记事本++7.3.1

记事本++7.3.1

好用且免费的代码编辑器

SublimeText3汉化版

SublimeText3汉化版

中文版,非常好用

禅工作室 13.0.1

禅工作室 13.0.1

功能强大的PHP集成开发环境

Dreamweaver CS6

Dreamweaver CS6

视觉化网页开发工具

SublimeText3 Mac版

SublimeText3 Mac版

神级代码编辑软件(SublimeText3)

vs code 可以在 Windows 8 中运行吗 vs code 可以在 Windows 8 中运行吗 Apr 15, 2025 pm 07:24 PM

VS Code可以在Windows 8上运行,但体验可能不佳。首先确保系统已更新到最新补丁,然后下载与系统架构匹配的VS Code安装包,按照提示安装。安装后,注意某些扩展程序可能与Windows 8不兼容,需要寻找替代扩展或在虚拟机中使用更新的Windows系统。安装必要的扩展,检查是否正常工作。尽管VS Code在Windows 8上可行,但建议升级到更新的Windows系统以获得更好的开发体验和安全保障。

vscode 扩展是否是恶意的 vscode 扩展是否是恶意的 Apr 15, 2025 pm 07:57 PM

VS Code 扩展存在恶意风险,例如隐藏恶意代码、利用漏洞、伪装成合法扩展。识别恶意扩展的方法包括:检查发布者、阅读评论、检查代码、谨慎安装。安全措施还包括:安全意识、良好习惯、定期更新和杀毒软件。

vscode怎么在终端运行程序 vscode怎么在终端运行程序 Apr 15, 2025 pm 06:42 PM

在 VS Code 中,可以通过以下步骤在终端运行程序:准备代码和打开集成终端确保代码目录与终端工作目录一致根据编程语言选择运行命令(如 Python 的 python your_file_name.py)检查是否成功运行并解决错误利用调试器提升调试效率

在PHP和Python之间进行选择:指南 在PHP和Python之间进行选择:指南 Apr 18, 2025 am 12:24 AM

PHP适合网页开发和快速原型开发,Python适用于数据科学和机器学习。1.PHP用于动态网页开发,语法简单,适合快速开发。2.Python语法简洁,适用于多领域,库生态系统强大。

visual studio code 可以用于 python 吗 visual studio code 可以用于 python 吗 Apr 15, 2025 pm 08:18 PM

VS Code 可用于编写 Python,并提供许多功能,使其成为开发 Python 应用程序的理想工具。它允许用户:安装 Python 扩展,以获得代码补全、语法高亮和调试等功能。使用调试器逐步跟踪代码,查找和修复错误。集成 Git,进行版本控制。使用代码格式化工具,保持代码一致性。使用 Linting 工具,提前发现潜在问题。

PHP和Python:解释了不同的范例 PHP和Python:解释了不同的范例 Apr 18, 2025 am 12:26 AM

PHP主要是过程式编程,但也支持面向对象编程(OOP);Python支持多种范式,包括OOP、函数式和过程式编程。PHP适合web开发,Python适用于多种应用,如数据分析和机器学习。

vscode 可以用于 mac 吗 vscode 可以用于 mac 吗 Apr 15, 2025 pm 07:36 PM

VS Code 可以在 Mac 上使用。它具有强大的扩展功能、Git 集成、终端和调试器,同时还提供了丰富的设置选项。但是,对于特别大型项目或专业性较强的开发,VS Code 可能会有性能或功能限制。

vscode 可以运行 ipynb 吗 vscode 可以运行 ipynb 吗 Apr 15, 2025 pm 07:30 PM

VS Code 运行 Jupyter Notebook 的关键是要确保 Python 环境正确配置,理解代码执行顺序与单元格顺序一致,并注意可能影响性能的大型文件或外部库。VS Code 提供的代码补全和调试功能可以大大提高编码效率和减少错误。

See all articles