python—day33 基于协程实现并发的套接字通信、网络IO操作、非阻塞IO模型、IO多路复用模型、异步IO；

基于协程实现并发的套接字通信：

　　server

from gevent import monkey,spawn;monkey.patch_all()
from threading import Thread
from socket import *

def talk(conn):
    while True:
        try:
            data=conn.recv(1024)
            if not data:break
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()
def server(ip,port,backlog=5):
    s = socket()
    s.bind((ip,port))
    s.listen(backlog)
    while True:
        conn, addr = s.accept()
        print(addr)
        # 通信
        g=spawn(talk,conn)
    s.close()

if __name__ == '__main__':
    spawn(server,'127.0.0.1',8080).join()
    # server(('127.0.0.1',8080))

　　client

from threading import Thread,current_thread
from socket import *
def client():
    client = socket()
    client.connect(('127.0.0.1', 8080))
    while True:
        data = '%s hello' % current_thread().name
        client.send(data.encode('utf-8'))
        res = client.recv(1024)
        print(res.decode('utf-8'))
if __name__ == '__main__':
    for i in range(1000):
        t=Thread(target=client)
        t.start()

 

网络IO操作：

　　server

from socket import *

s = socket()
s.bind(('127.0.0.1',8080))
s.listen(5)

while True:
    conn, addr = s.accept()
    print(addr)
    while True:
        try:
            data = conn.recv(1024)
            if not data: break
            print('from client msg: ',data)
        except ConnectionResetError:
            break
    conn.close()

 

　　client

from socket import *

client = socket()
client.connect(('127.0.0.1', 8080))

while True:

    data = input('>>: ').strip()
    if not data:continue
    client.send(data.encode('utf-8'))
    print('has send')

 

 

阻塞IO模型：blockingIOError!

 

阻塞IO模型：NonblockingIOError!

 

　　server

from socket import *
import time

s = socket()
s.bind(('127.0.0.1',8080))
s.listen(5)
s.setblocking(False)
r_list = []
w_list = []
while True:
    try:
        conn,addr = s.accept()
        r_list.append(conn)

    except BlockingIOError:
        # time.sleep(0.5)
        print('干其他活了！')

        print('rlist:',len(r_list))

        #收消息
        del_rlist = []
        for conn in r_list:
            try:
                data = conn.recv(1024)

                #linux 系统
                if not data:
                    conn.close()
                    del_rlist.append(conn)
                    continue

                # conn.send(data.upper())
                # w_list.append((conn,data.upper()))
            except BlockingIOError:
                continue

            except ConnectionResetError:
                conn.close()
                del_rlist.append(conn)


        #发消息
        del_wlist = []
        for item in w_list:
            try:
                conn= item[0]
                res= item[1]
                conn.send(res)
                del_wlist.append(item)
            except BlockingIOError:
                continue
            except ConnectionResetError:
                conn.close()
                del_wlist.append(conn)

        for conn in del_rlist:
            r_list.remove(conn)

        for item in del_wlist:
            w_list.remove(item)

 

　　client

from socket import *
import os
from threading import Thread

client = socket()
client.connect(('127.0.0.1',8080))

while True:
   data = '%s say hello' %os.getpid()
   client.send(data.encode('utf-8'))
   res = client.recv(1024)
   print(res.decode('utf-8'))

 

异步IO：

　　

from concurrent.futures import ThreadPoolExecutor
from threading import current_thread
import time
import os

def task(n):
    print('%s is running' %current_thread().name)
    time.sleep(2)
    return n**2

def parse(obj):
    res=obj.result()
    print(res)

if __name__ == '__main__':
    t=ThreadPoolExecutor(4)

    future1=t.submit(task,1)
    future1.add_done_callback(parse) #parse函数会在future1对应的任务执行完毕后自动执行，会把future1自动传给parse

    future2=t.submit(task,2)
    future2.add_done_callback(parse)

    future3=t.submit(task,3)
    future3.add_done_callback(parse)

    future4=t.submit(task,4)
    future4.add_done_callback(parse)

 

 

IO多路复用模型：multiplexing

 

　　server

 

from socket import *
import select

s = socket()
s.bind(('127.0.0.1',8080))
s.listen(5)
s.setblocking(False)
# print(s)

r_list=[s,]
w_list=[]
w_data={}
while True:
    print('被检测r_list： ',len(r_list))
    print('被检测w_list： ',len(w_list))
    rl,wl,xl=select.select(r_list,w_list,[],) #r_list=[server,conn]

    # print('rl: ',len(rl)) #rl=[conn,]
    # print('wl: ',len(wl))

    # 收消息
    for r in rl: #r=conn
        if r == s:
            conn,addr=r.accept()
            r_list.append(conn)
        else:
            try:
                data=r.recv(1024)
                if not data:
                    r.close()
                    r_list.remove(r)
                    continue
                # r.send(data.upper())
                w_list.append(r)
                w_data[r]=data.upper()
            except ConnectionResetError:
                r.close()
                r_list.remove(r)
                continue

    # 发消息
    for w in wl:
        w.send(w_data[w])
        w_list.remove(w)
        w_data.pop(w)

 

 

　　client

from socket import *
import os

client = socket()
client.connect(('127.0.0.1', 8080))

while True:
    data='%s say hello' %os.getpid()
    client.send(data.encode('utf-8'))
    res=client.recv(1024)
    print(res.decode('utf-8'))