GIL定时器线程queue进程池和线程池

Posted 2020-12-05 kingyanan

一、GIL
1、什么是GIL（这是Cpython解释器）
    GIL本质就是一把互斥锁，那既然是互斥锁，原理都一样，都是让多个并发线程同一时间只能
    有一个执行
    即：有了GIL的存在，同一进程内的多个线程同一时刻只能有一个在运行，意味着在Cpython中
        一个进程下的多个线程无法实现并行===》意味着无法利用多核优势
        但不影响并发的实现

    GIL可以被比喻成执行权限，同一进程下的所以线程 要想执行都需要先抢执行权限

2、为何要有GIL
    因为Cpython解释器自带垃圾回收机制不是线程安全的（对共享数据修改同时运行，不知谁改对了）

3、如何用

    01、GIL vs 自定义互斥锁
        GIL相当于执行权限,会在任务无法执行的情况，被强行释放
        自定义互斥锁即便是无法执行，也不会自动释放

        GIL能保护解释器级别代码（和垃圾回收机制有关）但保护不了其他共享数据（比如自己的代码）。
        所以在程序中对于需要保护的数据要自行加锁

    02、GIL的优缺点：
    优点：保证Cpython解释器内存管理的线程安全
    缺点：在Cpython解释器中，同一个进程下开启的多线程，同一时刻只能有一个线程执行，
          也就说Cpython解释器的多线程无法实现并行无法利用多核优势

    注意：
    a、GIL不能并行，但有可能并发，不一定为串行。因为串行是一个任务完完全全执行完毕后才进行下一个；
        而cpython中，一个线程在io时，被CPU释放时，会被强行取消GIL的使用权限
    b、多核（多CPU）的优势是提升运算效率
    c、计算密集型--》使用多进程，以用上多核
    d、IO密集型--》使用多线程

4、有两种并发解决方案：
    多进程：计算密集型
    多线程：IO密集型



 计算密集型:应该使用多进程
from multiprocessing import Process
from threading import Thread
import os,time
def work1():
    res=0
    for i in range(100000000):
        res*=i

def work2():
    res=0
    for i in range(100000000):
        res*=i

def work3():
    res=0
    for i in range(100000000):
        res*=i

def work4():
    res=0
    for i in range(100000000):
        res*=i

if __name__ == ‘__main__‘:
    l=[]
    # print(os.cpu_count()) #(计算cpu个数）本机为4核
    start=time.time()
    # p1=Process(target=work1) #
    # p2=Process(target=work2)
    # p3=Process(target=work3)
    # p4=Process(target=work4)

    p1=Thread(target=work1)
    p2=Thread(target=work2)
    p3=Thread(target=work3)
    p4=Thread(target=work4)

    p1.start()
    p2.start()
    p3.start()
    p4.start()
    p1.join()
    p2.join()
    p3.join()
    p4.join()
    stop=time.time()
    print(‘run time is %s‘ %(stop-start))


 IO密集型:应该使用多线程
from multiprocessing import Process
from threading import Thread
import os,time
def work1():
    time.sleep(5)

def work2():
    time.sleep(5)

def work3():
    time.sleep(5)

def work4():
    time.sleep(5)



if __name__ == ‘__main__‘:
    l=[]
    # print(os.cpu_count()) #本机为4核
    start=time.time()
    # p1=Process(target=work1) #
    # p2=Process(target=work2)
    # p3=Process(target=work3)
    # p4=Process(target=work4)

    p1=Thread(target=work1) #
    p2=Thread(target=work2)
    p3=Thread(target=work3)
    p4=Thread(target=work4)

    p1.start()
    p2.start()
    p3.start()
    p4.start()
    p1.join()
    p2.join()
    p3.join()
    p4.join()
    stop=time.time()
    print(‘run time is %s‘ %(stop-start))


二、定时器
    定时器，指定n秒后执行某操作
from threading import Timer,current_thread


def task(x):
    print(‘%s run....‘ %x)
    print(current_thread().name)


if __name__ == ‘__main__‘:
    t=Timer(3,task,args=(10,))
    t.start()
    print(‘主‘)

三、线程queue

import queue
队列：先进先出
q=queue.Queue(3)
q.put(1)
q.put(2)
q.put(3)

print(q.get())
print(q.get())
print(q.get())

 堆栈：先进后出
 q=queue.LifoQueue()
 q.put(1)
 q.put(2)
 q.put(3)
 print(q.get())
 print(q.get())
 print(q.get())

 优先级队列：优先级高先出来,数字越小，优先级越高
 q=queue.PriorityQueue()
 q.put((3,‘data1‘))
 q.put((-10,‘data2‘))
 q.put((11,‘data3‘))

 print(q.get())
 print(q.get())
 print(q.get())



三、基于多线程实现并发的套接字通信
服务端：
from socket import *
from threading import Thread

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

def server(ip,port,backlog=5):
    server = socket(AF_INET, SOCK_STREAM)
    server.bind((ip, port))
    server.listen(backlog)

    print(‘starting...‘)
    while True:
        conn, addr = server.accept()

        t = Thread(target=talk, args=(conn,))
        t.start()

if __name__ == ‘__main__‘:
    server(‘127.0.0.1‘,8080)


客户端：
from socket import *
import os

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

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





四、进程池与线程池

1、什么时候用池：
    池的功能是限制启动的进程数或线程数。

    什么时候应该限制？？？
    当并发的任务数远远超过了计算机的承受能力时，即无法一次性开启过多的进程数或线程数时
    就应该用池的概念将开启的进程数或线程数限制在计算机可承受的范围内

2、提交任务的两种方式：
    同步vs异步
    同步、异步指的是提交任务的两种方式

    同步：提交完任务后就在原地等待，直到任务运行完毕后拿到任务的返回值，再继续运行下一行代码
    异步：提交完任务（绑定一个回调函数）后根本就不在原地等待，直接运行下一行代码，等到任务有返回值后会自动触发回调函数

    程序的运行状态（阻塞，非阻塞）
    1、阻塞：
        IO阻塞
    2、非阻塞：
        运行
        就绪

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

#map(func, *iterables, timeout=None, chunksize=1)
取代for循环submit的操作

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

#result(timeout=None)
取得结果

#add_done_callback(fn)
回调函数



进程池：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import current_thread
import os,time,random

def task(n):
    print(‘%s is running‘%os.getpid())
    time.sleep(5)
    return n**2

def parse(future):
    time.sleep(1)
    res=future.result()
    print(‘%s 处理了 %s‘%(os.getpid(),res))

if __name__ == ‘__main__‘:
    pool=ProcessPoolExecutor(4)
    start=time.time()
    for i in range(1,5):
        future=pool.submit(task,i)       #异步提交任务
        future.add_done_callback(parse)  # parse会在future有返回值时立刻触发，并且将future当作参数传给parse
    pool.shutdown(wait=True)
    stop=time.time()
    print(‘主‘,os.getpid(),(stop-start))

‘‘‘
4340 is running
6572 is running
6652 is running
392 is running
5148 处理了 1
5148 处理了 4
5148 处理了 9
5148 处理了 16
主 5148 9.330533742904663
‘‘‘
最后由主进程一个一个处理结果



线程池：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import current_thread
import time,os,random

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

def parse(future):
    time.sleep(1)
    res=future.result()
    print(‘%s 处理了 %s‘%(current_thread().name,res))

if __name__ == ‘__main__‘:
    pool=ThreadPoolExecutor(4)
    start=time.time()
    for i in range(1,5):
        future=pool.submit(task,i)
        future.add_done_callback(parse)
    pool.shutdown(wait=True)
    stop=time.time()
    print(‘主‘,current_thread().name,(stop-start))

‘‘‘
ThreadPoolExecutor-0_0 is running
ThreadPoolExecutor-0_1 is running
ThreadPoolExecutor-0_2 is running
ThreadPoolExecutor-0_3 is running
ThreadPoolExecutor-0_2 处理了 9
ThreadPoolExecutor-0_1 处理了 4
ThreadPoolExecutor-0_3 处理了 16
ThreadPoolExecutor-0_0 处理了 1
主 MainThread 6.002343416213989

‘‘‘
最后谁（线程）空闲下来谁处理结果