进程池线程池 协程

Posted 2021-11-29 lakei

socket服务端实现并发

服务端需要满足以下3点：

　　1 固定的ip和port

　　2 24小时提供服务

　　3 能够实现并发

多线程实现并发：

服务端：

import socket
from  threading import Thread
import os

server  = socket.socket()
server.bind((‘127.0.0.1‘,8080))
server.listen(5)    #半连接池


def communicate(conn):
    while True:
        try:
            data= conn.recv(1024)
            if len(data)==0:break
            print(data)
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()


while True:
    conn, addr = server.accept()
    print(addr, os.getpid())
    t = Thread(target=communicate, args=(conn,))
    t.start()

服务端

客户端：

import socket

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

while True:
    info =input(">>>: ").encode(‘utf-8‘)
    if len(info)==0:continue
    client.send(info)
    data= client.recv(1024)
    print(data)

客户端

 

进程池 线程池

1 介绍

concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor ：线程池，提供异步调用
ProcessPoolExecutor：进程池，提供异步调用

2 基本方法
submit(fn,*args,**kwargs)
异步提交任务


shutdown(wait=Ture)
相当于进程池的pool.close() +pool.join()   操作
wait=True 等待池内所有任务执行完毕回收完资源后才继续
wait=False 立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit必须在shutdown之前

result
取得结果

add_done_callback(fn)
回调函数

from  concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
                                #线程池                #进程池
import time
import os

#实例化池对象
# 不知道参数的情况下，默认是当前计算机cpu个数乘以5 也可以指定线程个数

pool =ThreadPoolExecutor(5)  #创建一个池子，本机cpu个数4默认池子里面与20个线程

def task(n):
    print(n,os.getpid())
    time.sleep(2)
    return n**2

def call_back(n):
    print(‘我拿到了结果：%s‘%n.result())

‘‘‘
提交任务的方式:
    同步：提交任务之后，原地等待任务的返回结果，在继续执行下一行代码
    异步：提交任务之后，不等待任务的返回结果（通过回调函数拿到返回结果并处理），直接执行下一步操作
‘‘‘
#回调函数：异步提交之后一旦任务有返回结果，自动交给另一个去执行

t_list=[]
start=time.time()
for i in  range(20):

    n=pool.submit(task,i)   #提交任务
    t_list.append(n)
pool.shutdown() #关闭池子并且等待池子中所有的任务运行完毕
for n in t_list:
    print(‘>>>:‘,n.result())
print(‘主‘,time.time()-start)    #8.00445818901062


#
# pool= ProcessPoolExecutor(5)
#
# if __name__ == ‘__main__‘:
#     t_list=[]
#     start=time.time()
#     for i in range(20):
#         future= pool.submit(task,i).add_done_callback(call_back) 异步提交任务
#         t_list.append(future)
#
#     pool.shutdown()   #关闭池子并且的奶池子中所有的任务运行完毕
#
#     print(‘主‘,time.time()-start)    #8.328476428985596

 

协程

基于单线程来实现并发，只用一个主线程（很明显可利用的cpu只有一个）情况下实现并发，回顾并发的本质：切换+保存状态

　　CPU正在运行一个任务，会在两种情况下切走去执行其他的任务（切换由操作系统强制控制），一种情况是该任务发生了阻塞，另外一种情况是该任务计算的时间过长或有一个优先级更高的程序替代了它

　　一：其中第二种情况并不能提升效率，只是为了让cpu能够雨露均沾，实现看起来所有任务都被 ‘同时’ 执行的效果，如果多个任务都是纯计算的，这种切换反而会降低效率。yield本身就是一种在单线程下可以保存任务运行状态的方法yield可以保存任务运行状态。yield的状态保存与操作系统的保存线程状态很像，但是yield是代码级别控制的，更轻量级

    单线程实现并发
    在应用程序里控制多个任务的切换+保存状态
    优点：
        应用程序级别速度要远远高于操作系统的切换
    缺点；
        多个任务一旦有一个阻塞没有切，整个线程都阻塞在原地
        该线程内的其他的任务都不能执行了
    
        一旦引入协程，就需要检测单线程下所有的IO行为
        实现遇到IO就切换，少一个都不行，因为一旦一个任务阻塞了，整个线程就阻塞了
        其他的任务即便是可以计算，但是也无法运行了

　　切换+保存状态就一定能够提升程序的效率吗？
　　　　不一定
　　　　当你的任务是计算密集型，反而会降低效率
　　　　如果你的任务是io密集型，会提升效率

协程的目的：
    想要在单线程下实现并发
    并发指的是多个任务看起来是同时运行的
    并发=切换+保存状态

# import time
# def func1():
#     for i in  range(10000):
#         i+1
#
# def func2():
#     for i in range(10000):
#         i+1
#
# start = time.time()
#
# func1()     #串行
# func2()
# stop= time.time()
# print(stop-start)   #0.002000093460083008


# 基于yield并发执行
# import time
# def func1():
#     while True:
#         10000+1
#         yield
#
# def func2():
#     g=func1()
#     for  i in range(100):
#         time.sleep(0.1) #模拟io yield 并不会捕捉到并自动切换
#         i+1
#         next(g)
#
# start =time.time()
# func2()
# stop = time.time()
# print(stop-start)  #10.001571893692017

单纯切换反而降低运行效率

　　二：第一种情况的切换。在任务一遇到io情况下，切到任务二去执行，这样就可以利用任务一阻塞的时间完成任务二的计算，效率的提升就在于。

　　对于单线程下，我们不可避免程序中出现io操作，但如果我们能在自己的程序中（即用户程序级别，而非操作系统级别）控制单线程下的多个任务能在一个任务遇到io阻塞时就切换到另外一个任务计算，这样就保证了该线程能够最大限度地处于就绪态，即随时都可以被cpu执行的状态，相当于我们在用户程序级别将自己的io操作最大限度地隐藏起来，从而可以迷惑操作系统，让其看到：该线程好像是一直在计算，io比较少，从而更多的将cpu的执行权限分配给我们的线程。

　　协程的本质就是在单线程下，由用户自己控制一个任务遇到io阻塞了就切换另外一个任务去执行，以此来提升效率。为了实现它，我们需要寻找一种可以同时满足以下条件的解决方案：

    1 可以控制多个任务之间的切换，切换之前将任务的状态保存下来，以便重新运行时，可以基于暂停的位置继续执行。
    2 作为1的补充：可以检测io操作，在遇到io操作的情况下才发生切换

协程：协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。

1 python的线程属于内核级别的，即由操作系统控制调度（如单线程遇到io或执行时间过程就会被迫交出cpu执行权限，切换其他线程运行）
2 单线程内开启协程，一旦遇到io,就会从应用程序级别（而非操作系统）控制切换，以此来提升效率（！！！ 非io操作的切换与效率无关）

对比操作系统控制线程的切换，用户在单线程内控制协程的切换

优点:

　　1 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级

　　2 单线程内就可以实现并发的效果，最大限度地利用cpu

缺点：

　　1 协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程

　　2 协程指的是单个线程的并发，因而一旦协程出现阻塞，将会阻塞整个线程

总结协程：

　　1 必须在只有一个单线程里实现并发

　　2 修改共享数据不需要加锁

　　3 用户程序里自己保存多个控制流的上下文

　　4 一个协程遇到IO操作自动切换到其他协程（gevent模块）

 

Gevent模块

#用法
g1=gevent.spawn(func,1,,2,3,x=4,y=5)创建一个协程对象g1，spawn括号内第一个参数是函数名，如eat，后面可以有多个参数，可以是位置实参或关键字实参，都是传给函数eat的

g2=gevent.spawn(func2)

g1.join() #等待g1结束

g2.join() #等待g2结束

#或者上述两步合作一步：gevent.joinall([g1,g2])

g1.value#拿到func1的返回值




遇到IO阻塞时会自动切换任务

from gevent import monkey;monkey.patch_all()  # 监测代码中所有io行为

from gevent import spawn
# gevent 本身识别不了time.sleep等不属于该模块内的io操作
import time

def heng(name):
    print(‘%s 哼哼哼‘ % name)
    time.sleep(2)
    print(‘%s 哼‘ % name)

def ha(name):
    print(‘%s 哈哈哈哈‘ % name)
    time.sleep(3)
    print(‘%s 哈‘ % name)
start= time.time()
s1=spawn(heng,‘egon‘)
s2 = spawn(ha,‘kwevin‘)
s1.join()
s2.join()

print(‘主‘,time.time()-start)
# egon 哼哼哼
# kwevin 哈哈哈哈
# egon 哼
# kwevin 哈
# 主 3.003171682357788 3秒多一点 多出来的是cpu来回切换的时间

通过gevent实现单线程下的socket并发

from  gevent import monkey;monkey.patch_all()
from gevent import spawn
import socket

def communicate(conn):
    while True:
        try:
            data=conn.recv(1024)
            if len(data)==0:break
            print(data.decode(‘utf-8‘))
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()

def server():
    server =socket.socket()
    server.bind((‘127.0.0.1‘,8081))
    server.listen(5)
    while True:
        conn,addr =server.accept()
        spawn(communicate,conn)

if __name__ == ‘__main__‘:
    s1=spawn(server)
    s1.join()

服务端

from  threading import  Thread,current_thread
import socket

def client():
    client= socket.socket()
    client.connect((‘127.0.0.1‘,8081))
    n=1
    while True:
        data=‘%s%s ‘%(current_thread().name,n)
        n+=1
        client.send(data.encode(‘utf-8‘))
        info=client.recv(1024)
        print(info)

if __name__ == ‘__main__‘:
    for i in range(100):
        t=Thread(target=client)
        t.start()

多线程并发多个客户端