一.twisted-tcp1

(一)server.py

# coding:utf8
"""
1 (多进程多线程多协程)并发问题:死锁(等待与唤醒)
  并发解决(资源切换 系统调度 同步与等待)
2 阻塞IO等待
 （1)服务端与客户端已连接 服务端recv等待客户端发送
  (2)客户端recv
  非阻塞IO(异步操作处理): 单独定期轮询问制度(烧开水例子)
3 twisted: 外部事件发生时使用回调机制来触发相应的处理操作
  多个任务在一个线程中执行(不再关注线程安全)
  所有处理事件reactor以单线程模式持续运行(注册注销运行回调处理等)
  当需要回调则暂循 完成回调后继续循环
4 所有输入注册->Reactor->管理所有输出

pip install twisted
先装pip install service-identity(否则警告提示)
TCP 定义服务器类  工厂类(统一管理服务器)  reactor调用工厂类对象
"""

import twisted.internet.protocol, twisted.internet.reactor

class Twisted_TcpServer_Protocol(twisted.internet.protocol.Protocol):
    def connectionMade(self):  # 重写方法 客户端连接时触发
        server_addr=self.transport.getPeer().host
        client_addr=self.transport.getPeer()
        print(f"【客户端】{server_addr} : {client_addr}")

    def dataReceived(self, data:bytes): #监听接收数据到放在data
        "接收客户端数据 轮询的地方(等待客户端发送) 服务器在此方法可发送"
        client_addr=self.transport.getPeer()
        print(f"【接收数据】:{data.decode('utf-8')}",client_addr)
        send_binary_data=bytes("【ECHO】","utf8")+data
        self.transport.write( send_binary_data ) #灰底提示

class DefaultServerFactory(twisted.internet.protocol.Factory):
    protocol=Twisted_TcpServer_Protocol #注册回调操作 放到工厂 再放入reactor

if __name__ == '__main__': ## listenTCP代表服务器端
    twisted.internet.reactor.listenTCP(9999,DefaultServerFactory())
    print("【服务器】监听9999 启动完毕，等待客户端连接...")
    twisted.internet.reactor.run()
    #事件轮询 【测试】此处两个ip均成功 127.0.0.1:9999  192.168.14.64:9999

(二) client.py

# coding:utf8
"""
pycharm_Run/Debug Configuration _勾选Allow prallel run
(同时 运行多客户端程序)
"""

from twisted.internet import protocol, reactor
HOST,PORT = 'localhost',9999

class Twisted_Tcp_Client_Protocol(protocol.Protocol):
    "该类继承和接受其所有属性和方法 protocol.Protocol类"
    def senData(self): #用户自定义方法
        data = input('>').strip().encode()
        if data:
            print('【sending】%s' % str(data, 'utf8'))
            self.transport.write(data)
        else:
            print("没有输入内容,客户端程序自动退出")
            self.transport.loseConnection()

    def connectionMade(self): #重写
        print("连接服务器成功!!",self.transport.getPeer())
        self.senData()

    def dataReceived(self, data):#重写
        print("服务器的信息:",str(data,"utf8"))
        hostAddr=self.transport.getHost()
        myAddr=self.transport.getPeer()
        print( hostAddr )
        print( myAddr )
        self.senData()

class Twisted_Tcp_Client_Factory(protocol.ClientFactory):
    protocol = Twisted_Tcp_Client_Protocol #重写类属性?定义回调
    clientConnectionLost = clientConnectionFailed = \
        lambda self, connector, reason: reactor.stop()
    """
    lambda 传入三个参数给无名函数 返回值是reactor.stop()
    疑reactor.stop函数体内有该三个参数
    """

if __name__ == '__main__': ## 代表客户端connectTCP
    reactor.connectTCP(HOST, PORT,Twisted_Tcp_Client_Factory())
    reactor.run()

二.twisted-udp

# coding:utf8

from twisted.internet import protocol,reactor
SEVER_PORT=9999
class Twisted_udp_server(protocol.DatagramProtocol):
    def datagramReceived(self,datagram:bytes , addr:tuple):
        print("【服务器】监听到客户端连接: ip=%s port=%s"%addr)
        print(f"其消息={datagram.decode()}")
        message_bytes= "【ECHO】".encode() + datagram
        self.transport.write( message_bytes , addr)

if __name__ == '__main__':
    reactor.listenUDP( SEVER_PORT,Twisted_udp_server())
    print('\n--- UOPserver is waiting for connection ---')
    reactor.run()

# coding:utf8
from twisted.internet import protocol, reactor

SERVER_HOST,SERVER_PORT,= '127.0.0.1',9999
CLIENT_PORT=0
class UDPClient(protocol.DatagramProtocol):
    def senData(self):
        data = input('>').encode()
        if data:
            print('[sending]%s' % str(data, 'utf8'))
            self.transport.write( data )
        else:
            print("[sending]发送数据为空,程序退出")
            reactor.stop()

    def startProtocol(self): #代表是客户端的标记1
        self.transport.connect(SERVER_HOST,SERVER_PORT)
        print("[连接成功]可发送接收",self.transport.getHost())
        self.senData()

    def datagramReceived(self, datagram, addr):
        print("[接收]%s : %s"%(addr,datagram.decode()))
        self.senData()

if __name__ == '__main__': #代表是客户端的标记2
    reactor.listenUDP(CLIENT_PORT, UDPClient())
    reactor.run()

三.twisted-defer

# coding:utf8
import sys
import time

from twisted.internet import reactor #进入源码

def f0():
    "进入reactor源码 pycharm "
    for i,(k,v) in enumerate(sys.modules.items()):
        print(f"[{i+1}]",k)
        print(v)

def f1():
    def main(message:str,count:int):
        print(message,count)
        reactor.stop()
        print("reactor停止")

    reactor.callWhenRunning(main, "main函数及多参",100)
    reactor.run()

def f2(): # 过1秒时只执行1次 递归变成无限循环
    def fun(k1, k2):
        t=time.strftime("%Y-%m-%d %X")
        k1,k2=1+k1,1+k2
        print(f"[{t}]调用fun 参数=%s %s" % (k1, k2))
        reactor.callLater(1, fun, k1, k2) #只执行1次

    def main(message):
        print(message)
        k1, k2 = 10, 20
        reactor.callLater(1, fun, k1, k2)#只执行1次

    reactor.callWhenRunning(main, "twisted开始运行")
    reactor.run()

if __name__ == '__main__':
    pass
    #f1()
    f2()

# coding:utf8
import time #defer推迟 reacter(核)反应器(堆)
from twisted.internet import defer,reactor

class DeferHandle:
    def __init__(self):
        self.defer=defer.Deferred()
    def get_defer(self):#让外部获取defer对象
        return self.defer
    def work(self):
        print("假设2秒用于完成下载功能(阻塞状态?)...")
        time.sleep(2)
        self.defer.callback("finish") #执行回调
    def handle_success(self,result):#result="finish"
        print("已成功完成 结果:%s"%result)
    def handle_error(self,exp):
        print("处理报错:%s"%exp)

def stop():
    reactor.stop()
    print("服务调用结束。。。")
def main():
    defer_client = c = DeferHandle()
    reactor.callWhenRunning( defer_client.work  )
    c.get_defer().addCallback( c.handle_success )
    c.get_defer().addErrback(   c.handle_error  )
    reactor.callLater( 3,stop ) # 加了延迟时间
    reactor.run()

if __name__ == '__main__':
    main()

(一)server.py

# coding:utf8
from twisted.internet import protocol,reactor

class TcpServer(protocol.Protocol):
    def connectionMade(self): #有客户端连接时触发
        hostAddr=self.transport.getHost().host
        myAddr=self.transport.getPeer()
        print(f"【服务器】地址{hostAddr}:{myAddr}")

    def dataReceived(self, data:bytes):
        # 接收客户端数据 轮询的地方(等待客户端发送)
        myAddr=self.transport.getPeer()#data.decode()
        print(f"【服务器】接收:{str(data,'utf8')}",myAddr)
        send_binary_data=bytes("【ECHO】","utf8")+data
        self.transport.write( send_binary_data ) #灰底提示

class DefaultServerFactory(protocol.Factory):
    protocol=TcpServer#注册回调->放到工厂->再放入reactor

if __name__ == '__main__':  #服务器标记 有"监听"功能
    reactor.listenTCP(9999,DefaultServerFactory())
    m= "【浏览器测试1】127.0.0.1:9999 localhost:9999\n"
    m+="【浏览器测试2】192.168.14.64:9999"
    print(f"服务器启动完毕，事件轮询等待客户端连接\n{m}")
    reactor.run() #灰底提示

(二) client.py

# coding:utf8
# twisted_deferred_tcp_echo_client.py
import time
from twisted.internet import reactor, threads, protocol

class DeferClient(protocol.Protocol):
    def connectionMade(self):  # 重写 只执行一次
        hostAddr = self.transport.getHost().host
        myAddr = self.transport.getPeer()
        print("【客户端】已连服务器地址", hostAddr, myAddr)
        self.send()

    def send(self):
        m = "请输入要发送数据(结束通讯空值直接回车):> "
        input_data = input(m).strip().encode()
        if input_data:
            self.transport.write(input_data)
        else: # reactor.stop()
            self.transport.loseConnection()

    def dataReceived(self, data: bytes):  # 重写
        content = data.decode()

        threads.deferToThread(
            self.handle_request, content
        ).addCallback( self.handle_success )

    def handle_request(self, content):
        "1大文件处理需要时间处理"
        m='【客户端】1.对服务器端发来数据处理 此处产生1秒延迟..'
        print(m, content)
        time.sleep(1)
        return content

    def handle_success(self, result):
        "2 处理完毕的通知回调 如果处理错误还可以添加 错误通知"
        print("2.处理完毕,接收参数=", result)
        self.send()

class DefaultClientFactory(protocol.ClientFactory):
    protocol = DeferClient
    clientConnectionLost = clientConnectionFailed = \
        lambda self, connector, reason: reactor.stop()

if __name__ == '__main__':
    reactor.connectTCP(
        "localhost", 9999, DefaultClientFactory())
    reactor.run()