并发编程之io模型(代码片段)

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcong
from socket import *

def talk():
    client = socket(AF_INET,SOCK_STREAM)
    client.connect((\'127.0.0.1\',8806))

    while True:
        mes = input(\'>>>:\').strip()
        if not mes:continue
        client.send(mes.encode(\'utf-8\'))
        data = client.recv(1024)
        print(data.decode(\'utf-8\'))
    client.close()

if __name__ == \'__main__\':
    talk()

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcong

from socket import *
from threading import Thread,currentThread

def talk(conn):
    while True:
        try:
            data = conn.recv(1024)
            if not data:break
            print(data.decode(\'utf-8\'))
            conn.send((\'%s hello\'%currentThread().getName()).encode(\'utf-8\'))
        except ConnectionResetError:
            break
    conn.close()

def server(ip,port):
    server = socket(AF_INET,SOCK_STREAM)
    server.bind((ip,port))
    server.listen(5)
    while True:
        print(\'staring...\')
        conn,addres = server.accept()
        print(addres)
        t = Thread(target=talk,args=(conn,))
        t.start()

    server.close()

if __name__ == \'__main__\':
    server(\'127.0.0.1\',8806)

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcong

from socket import *

client = socket(AF_INET,SOCK_STREAM)
client.connect((\'127.0.0.1\',6666))
while True:
    msg = input(\'>>>:\').strip()
    if not msg:continue
    client.send(msg.encode(\'utf-8\'))
    data = client.recv(1024)
    print(data.decode(\'utf-8\'))
client.close()

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcong

# 非阻塞IO --> 单进程下多线程速度极快，当前程序效率很高（一直占用CPU），但影响其它程序的执行
\'\'\'
  非阻塞的recv系统调用调用之后，进程并没有被阻塞，内核马上返回给进程，如果数据还没准备好，
此时会返回一个error。进程在返回之后，可以干点别的事情，然后再发起recv系统调用。重复上面的过程，
循环往复的进行recv系统调用。这个过程通常被称之为轮询。轮询检查内核数据，直到数据准备好，再拷贝数据到进程，
进行数据处理。需要注意，拷贝数据整个过程，进程仍然是属于阻塞的状态。

\'\'\'
from socket import *
server = socket(AF_INET,SOCK_STREAM)
server.bind((\'127.0.0.1\',6666))
server.listen(5)
server.setblocking(False) # 不阻塞，默认是阻塞
rlist = [] # 存放链接
slist = [] # 存放消息
print(\'staring...\')
while True:
    try:
        conn,addres = server.accept() # IO堵塞
        rlist.append(conn)
        print(rlist)
    except BlockingIOError: # 未收到链接抛异常
        #print(\'数据未准备好！\')
        # 收消息
        del_rlist = []
        for conn in rlist: # 遍历链接列表
            try:
                data  = conn.recv(1024)
                if not data:
                    continue
                #conn.send(data.upper())
                slist.append((conn,data))
            except BlockingIOError:  # 碰到IO阻塞
                continue
            except Exception:
                conn.close()
                del_rlist.append(conn)
        # 发消息
        del_slist = []
        for item in slist:
            try:
                conn = item[0]
                data = item[1]
                conn.send(data.upper()) # 可能会抛异常，即IO阻塞
                del_slist.append(item) # 没抛异常，就将发成功的信息加到将要被删除的队列中
            except BlockingIOError: # 未发送成功
                continue
        for item in del_slist: # 遍历列表，将其中存放的已发送成功的信息删除
            slist.remove(item)
        for conn in del_rlist: # 遍历列表，将未收到数据的链接删除
            rlist.remove(conn)

server.close()

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcong
from socket import *


client = socket(AF_INET,SOCK_STREAM)
client.connect((\'127.0.0.1\',6868))

while True:
    mes = input(\'>>>:\').strip()
    if not mes:continue
    client.send(mes.encode(\'utf-8\'))
    data = client.recv(1024)
    print(data.decode(\'utf-8\'))
client.close()

#！/usr/bin/env python3
#-*- coding:utf-8 -*-
# write by congcong

# 多路复用IO,可以同时检测多个IO,而非阻塞IO只能检测一个IO,单个链接时非阻塞IO效率高，多个链接时多路复用更佳
\'\'\'
    select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。
它的基本原理就是select/epoll这个function会不断的轮询所负责的所有socket，
当某个socket有数据到达了，就通知用户进程。
\'\'\'
from socket import *
import select

server = socket(AF_INET,SOCK_STREAM)
server.bind((\'127.0.0.1\',6868))
server.listen(5)
server.setblocking(False) # 不阻塞
rlist = [server,] # 存放链接和套接字(server,conn)
wlist = [] # 存放发送消息的套接字()
wdata =  
while True:
    rl,wl,xl = select.select(rlist,wlist, [], 0.5)  # 后两个参数分别表示异常列表和超时时间
    print(wl)
    # 收消息
    for sock in rl:
        if sock == server: # 收到server
            conn,addres = sock.accept()
            rlist.append(conn) # 加入列表
        else: # 即收到的是 conn
            try:
                data = sock.recv(1024)
                if not data:
                    sock.close()
                    rlist.remove(sock) # 针对linux系统报错，一直接收的特点
                wlist.append(sock)
                wdata[sock] = data.upper()
            except Exception: # 关闭未收到数据的无用链接和删除套接字
                sock.close()
                rlist.remove(sock)
    # 发消息
    for sock in wl:
        data = wdata[sock] # 获取字典套接字对应的数据
        sock.send(data) # 发送数据
        wlist.remove(sock) # 删除已经接收的套接字
        wdata.pop(sock) # 删除已经发送成功的数据
server.close()