并发编程—线程

目录

• 线程总结
  • 线程基本概念
  • GIL锁
  • threading模块中的Thread模块
  • 线程中的其他方法
  • 守护线程
  • 线程锁
    • 互斥锁
    • 递归锁
      • 死锁现象
  • 线程队列
  • 线程池/进程池
    • 概念
    • 特点
    • 方法

线程总结

线程基本概念

• 线程是能被cpu(操作系统)调度的最小单位

• 特点：数据共享，数据不安全，操作系统级别，能使用多核，开启、关闭时间花销比进程小几百倍

• 线程节省的是i/o操作的时间，而不是cpu计算时间，因为cpu的计算速度速度非常快，大部分情况下，我们没有办法将一条进程中的所有io操作都规避掉

GIL锁

• GIL（the Global Interpreter Lock）全局解释器锁

  • cpython解释器中的gc垃圾回收机制（引用计数+分代回收）
  • 在cpython/pypy解释器中，不能使用多核，jpython可以使用多核

• GIL会导致同一个进程中的多个线程只有一个线程能真正被cpu执行

threading模块中的Thread模块

import time
from threading import Thread

def fun(i):
    print('start%s'%i)
    time.sleep(1)
    print('end%s'%i)
if __name__ == '__main__':
    t_ls=[]
    for i in range(10):
        t=Thread(target=fun,args=(i,))
        t.start()
        t_ls.append(t)
    for t in t_ls:t.join()
    print('执行结束')

线程中的其他方法

• current_thread().ident:查看当前的线程id
• .enumerate():查看当前存活的线程，包括主线程，结果为一个对象
• .active_count():查看当前存活线程的数量

import os
import time
from threading import Thread,current_thread,enumerate,active_count
# from multiprocessing import Process as Thread
def func(i):
    print('start%s'%i,current_thread().ident)
    time.sleep(1)
    print('end%s'%i)
if __name__ == '__main__':
    tl = []
    for i in range(10):
        t = Thread(target=func,args=(i,))
        t.start()
        tl.append(t)
        print(t.ident, os.getpid())
    print(enumerate(),active_count())
    for t in tl:t.join()
    print('所有的线程都执行完了')

守护线程

• 守护线程不同于守护进程，守护线程是指被守护的子线程直到主线程结束之后才结束，而守护进程随着主代码的的结束就结束了
• 守护进程与守护线程为什么会有区别
  • 第一，结束原理不同
  • 所有的进程，线程都得需要主线程来回收资源
• 主线程结束-->守护线程结束-->子进程结束-->主进程结束-->主进程回收资源

import time
from threading import Thread

def son():
    while True:
        print('in son')
        time.sleep(1)

def son2():
    for i in range(3):
        print('in son2 ****')
        time.sleep(1)

t = Thread(target=son)
t.daemon = True
t.start()
Thread(target=son2).start()

线程锁

• 由于线程之间数据共享，所以线程存在数据不安全现象，可以通过加锁解决
• 一般的赋值操作（-=、+=、*=、/=），if、while等操作都会产生数据不安全现象，而字典，列表的方法操作数据是安全的（我们可以通过dis.dis模块去查看底层的语法从而分析为什么数据不安全），所以我们尽量不要操作全局变量或者类中的静态变量

#赋值操作
import dis
a = 0
def func():
    global a
    a += 1
dis.dis(func)
'''
 56           0 LOAD_GLOBAL              0 (a)
              2 LOAD_CONST               1 (1)
              4 INPLACE_ADD
              # GIL锁切换了
              6 STORE_GLOBAL             0 (a)
'''

#列表的append方法
import dis
a = []
def func():
    a.append(1)

dis.dis(func)
'''
 70           0 LOAD_GLOBAL              0 (a)
              2 LOAD_ATTR                1 (append)
              4 LOAD_CONST               1 (1)
              6 CALL_FUNCTION            1
              8 POP_TOP
'''

• 线程中的锁有互斥锁和递归锁

互斥锁

#互斥锁
from threading import Thread,Lock
n = 0
def add(lock):
    for i in range(500000):
        global n
        with lock:								#涉及到赋值操作，加互斥锁
            n += 1
def sub(lock):
    for i in range(500000):
        global n
        with lock:								#涉及到赋值操作，加互斥锁
            n -= 1

t_l = []
lock = Lock()
for i in range(2):
    t1 = Thread(target=add,args=(lock,))
    t1.start()
    t2 = Thread(target=sub,args=(lock,))
    t2.start()
    t_l.append(t1)
    t_l.append(t2)
for t in t_l:
    t.join()
print(n)

递归锁

死锁现象

• 死锁现象是由于两个或两个以上的进程/线程由于相互争夺资源而产生相互等待的现象
• 解决死锁现象最快速的办法就是讲多把互斥锁改成递归锁，但这样的效率会变低

import time
from threading import Thread,Lock,RLock
fork_lock = noodle_lock = RLock()
# fork_lock = RLock()

def eat(name):
    noodle_lock.acquire()
    print(name,'抢到面了')
    fork_lock.acquire()
    print(name, '抢到叉子了')
    print(name,'吃面')
    time.sleep(0.1)
    fork_lock.release()
    print(name, '放下叉子了')
    noodle_lock.release()
    print(name, '放下面了')

def eat2(name):
    fork_lock.acquire()
    print(name, '抢到叉子了')
    noodle_lock.acquire()
    print(name,'抢到面了')
    print(name,'吃面')
    noodle_lock.release()
    print(name, '放下面了')
    fork_lock.release()
    print(name, '放下叉子了')

Thread(target=eat,args=('团团',)).start()
Thread(target=eat2,args=('圆圆',)).start()
Thread(target=eat,args=('欢欢',)).start()
Thread(target=eat2,args=('喜喜',)).start()

线程队列

• 线程队列有三种
  • Queue：先进先出队列，适用于服务类场景
  • LifoQueue：后进先出队列，适用于算法等场景
  • PriorityQueue：优先级队列，适用于服务类场景，比如会员，vip购票等

#Queue
import queue   				# 线程之间数据安全的容器队列
q = queue.Queue(4)   		# fifo 先进先出的队列
q.put(1)
q.put(2)
q.put(3)
q.put(4)
print('4 done')
# q.put_nowait(5)
print('5 done')
try:
	while True:
    	print(q.get_nowait())
except queue.Empty:pass
	print('队列为空,继续其他内容')
	
#LifoQueue
from queue import LifoQueue   # last in first out 后进先出 栈
lq = LifoQueue()
lq.put(1)
lq.put(2)
lq.put(3)
print(lq.get())
print(lq.get())
print(lq.get())

#priorityQueue
from queue import PriorityQueue  # 优先级队列
priq = PriorityQueue()
priq.put((2,'团团'))
priq.put((1,'圆圆'))
priq.put((0,'乐乐'))

print(priq.get())
print(priq.get())
print(priq.get())

线程池/进程池

概念

• 是指在程序开始但还没有提交任务的时候会创建几个线程或者进程放在池子里，存放线程的就叫线程池，存放进程的就叫进程池

特点

• 当有任务来临时，可以直接使用池中的进程或者线程，而且池中的进程或者线程一直都会在池中，可以反复使用，这样就大大减少了开启、关闭、调度进程/线程所需要的时间开销
• 池中的线程/进程个数控制了操作系统需要调度的任务个数，这样有利于提高操作系统的效率，减轻操作系统的负担

方法

#submit函数
import os
import time,random
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
def func(a,b):
    print(os.getpid(),'start',a,b)
    time.sleep(random.randint(1,4))
    print(os.getpid(),'end')
    return a*b

if __name__ == '__main__':
    tp = ProcessPoolExecutor(4)
    futrue_l = {}
    for i in range(20):          # 异步非阻塞的
        ret = tp.submit(func,i,b=i+1)
        futrue_l[i] = ret
        # print(ret.result())    # Future未来对象
    for key in futrue_l:         # 同步阻塞的
        print(key,futrue_l[key].result())
        
# map  只适合传递简单的参数,并且必须是一个可迭代的类型作为参数
import os
import time,random
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
def func(a):
    print(os.getpid(),'start',a[0],a[1])
    time.sleep(random.randint(1,4))
    print(os.getpid(),'end')
    return a[0]*a[1]

if __name__ == '__main__':
    tp = ProcessPoolExecutor(4)
    ret = tp.map(func,((i,i+1) for i in range(20)))
    for key in ret:       # 同步阻塞的
        print(key)
        
#回调函数  add_done_callback(函数名)
import time,random
from threading import current_thread
from concurrent.futures import ThreadPoolExecutor

def func(a,b):
    print(current_thread().ident,'start',a,b)
    time.sleep(random.randint(1,4))
    print(current_thread().ident,'end',a)
    return (a,a*b)

def print_func(ret):       					# 异步阻塞
    print(ret.result())

if __name__ == '__main__':
    tp = ThreadPoolExecutor(4)
    futrue_l = {}
    for i in range(20):        		 		# 异步非阻塞的
        ret = tp.submit(func,i,b=i+1)
        ret.add_done_callback(print_func)   # ret这个任务会在执行完毕的瞬间立即触发print_func函数,并且把任务的返回值对象传递到print_func做参数
        # 异步阻塞 回调函数 给ret对象绑定一个回调函数,等待ret对应的任务有了结果之后立即调用print_func这个函数
        # 就可以对结果立即进行处理,而不用按照顺序接收结果处理结果