以下程序均来自《Python.UNIX和Linux系统管理指南》

用twisted实现检测tcp端口

twisted_check_tcp_port.py#!/usr/bin/env pythonfrom twisted.internet import reactor, protocolimport sysclass PortCheckerProtocol(protocol.Protocol):        def __init__(self):                print "Created a new protocol"        def connectionMade(self):                print "Connection made"                reactor.stop()class PortCheckerClientFactory(protocol.ClientFactory):        protocol = PortCheckerProtocol        def clientConnectionFailed(self, connector, reason):                print "Connection failed because", reason                reactor.stop()if __name__ == '__main__':        host, port = sys.argv[1].split(':')        factory = PortCheckerClientFactory()        print "Testing %s" % sys.argv[1]        reactor.connectTCP(host, int(port), factory)        reactor.run()

运行结果：

[root@centos python]# python twisted_check_tcp_port.py 127.0.0.1:80

Testing 127.0.0.1:80

Created a new protocol

Connection made

[root@centos python]# python twisted_check_tcp_port.py 127.0.0.1:8080

Testing 127.0.0.1:8080

Connection failed because [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.error.ConnectionRefusedError'>: Connection was refused by other side: 111: Connection refused.

]

用twisted实现perspective broker

server端

twisted_perspectiv_broker.py#!/usr/bin/env pythonimport osfrom twisted.spread import pbfrom twisted.internet import reactorclass PBDirLister(pb.Root):        def remote_ls(self, directory):                try:                        return os.listdir(directory)                except OSError:                        return []        def remote_ls_boom(self, directory):                return os.listdir(directory)if __name__ == '__main__':        reactor.listenTCP(9876, pb.PBServerFactory(PBDirLister()))        reactor.run()

client端

twisted_perspectiv_broker_client.py#!/usr/bin/python envfrom twisted.spread import pbfrom twisted.internet import reactordef handle_err(reason):        print "an error occurred", reason        reactor,stop()def call_ls(def_call_obj):        return def_call_obj.callRemote('ls', '/usr')def print_ls(print_result):        print print_result        reactor.stop()if __name__ == '__main__':        factory = pb.PBClientFactory()        reactor.connectTCP('localhost', 9876, factory)        d = factory.getRootObject()        d.addCallback(call_ls)        d.addCallback(print_ls)        d.addErrback(handle_err)        reactor.run()

其中def_call_obj.callRemote('ls', '/usr') ls也可以换成ls_boom

运行结果：

运行服务端（如果正确，服务端不会打印任何东西）

[root@centos python]# python twisted_perspectiv_broker.py

运行客户端（正确运行会列出所要列出的目录，如果没有该目录则返回一个空列表）

[root@centos python]# python twisted_perspectiv_broker_client.py

['share', 'libexec', 'games', 'tmp', 'etc', 'lib', 'sbin', 'X11R6', 'kerberos', 'src', 'include', 'local', 'bin']