python从入门到放弃--线程进阶

# ### 死锁,递归锁,互斥锁

from threading import Thread,Lock
import time
noodle_lock = Lock()
kuaizi_lock = Lock()

def eat1(name):
    noodle_lock.acquire()
    print("%s 拿到面条" % (name))
    kuaizi_lock.acquire()
    print("%s 拿到筷子" % (name))
    
    print("开始吃面条 ... ")
    time.sleep(0.5)
    
    kuaizi_lock.release()
    print("%s 放下筷子" % (name))
    noodle_lock.release()
    print("%s 放下面条" % (name))

def eat2(name):
    kuaizi_lock.acquire()
    print("%s 拿到筷子" % (name))
    noodle_lock.acquire()
    print("%s 拿到面条" % (name))

    print("开始吃面条 ... ")
    time.sleep(0.5)
    noodle_lock.release()
    print("%s 放下面条" % (name))
    kuaizi_lock.release()
    print("%s 放下筷子" % (name))

if __name__ == "__main__":
    name_lst1 = ["周永玲","张龙"]
    name_lst2 = ["李诗韵","黄乐锡"]

    for name in name_lst1:
        Thread(target=eat1,args=(name,)).start()
        
    for name in name_lst2:
        Thread(target=eat2,args=(name,)).start()

可以看到线程是异步并发的执行的，也就说当线程异步并发时会造成一人一把锁从而造成锁数不匹配从而造成死锁现象

# 死锁 : 连续上锁,不解锁
"""
lock.acquire()
lock.acquire()
lock.release()
lock.release()
print("正常运行2")
"""

(2) 递归锁

from threading import Thread,RLock,Lock
作用:
递归锁专门用来解决死锁现象,为了应急
临时解决线上死锁问题,使之正常运行;
语法:
rlock = RLock()

rlock = RLock()
def func():    
    rlock.acquire()
    rlock.acquire()
    rlock.acquire()
    rlock.acquire()
    rlock.acquire()
    
    rlock.release()
    rlock.release()
    rlock.release()
    rlock.release()
    rlock.release()
    print("正常运行")

func()

lock = Lock()

(3) 解决吃面条的死锁问题

noodle_lock = kuaizi_lock = RLock()
def eat1(name):
    noodle_lock.acquire()
    print("%s 拿到面条" % (name))
    kuaizi_lock.acquire()
    print("%s 拿到筷子" % (name))
    
    print("开始吃面条 ... ")
    time.sleep(0.5)
    
    kuaizi_lock.release()
    print("%s 放下筷子" % (name))
    noodle_lock.release()
    print("%s 放下面条" % (name))

def eat2(name):
    kuaizi_lock.acquire()
    print("%s 拿到筷子" % (name))
    noodle_lock.acquire()
    print("%s 拿到面条" % (name))

    print("开始吃面条 ... ")
    time.sleep(0.5)
    noodle_lock.release()
    print("%s 放下面条" % (name))
    kuaizi_lock.release()
    print("%s 放下筷子" % (name))

if __name__ == "__main__":
    name_lst1 = ["周永玲","张龙"]
    name_lst2 = ["李诗韵","黄乐锡"]

    for name in name_lst1:
        Thread(target=eat1,args=(name,)).start()
        
    for name in name_lst2:
        Thread(target=eat2,args=(name,)).start()

使用递归锁之后会让多线程变为同步的的执行顺序，而一般不推荐使用递归锁，出现死锁的现象大部分情况是因为前期逻辑有问题而导致的，

(4) 正确使用互斥锁: 尽量不要设计锁嵌套,容易死锁

mylock = Lock()
def eat1(name):
    mylock.acquire()
    print("%s 拿到面条" % (name))
    print("%s 拿到筷子" % (name))
    
    print("开始吃面条 ... ")
    time.sleep(0.5)    

    print("%s 放下筷子" % (name))
    print("%s 放下面条" % (name))
    mylock.release()

def eat2(name):
    mylock.acquire()
    print("%s 拿到筷子" % (name))    
    print("%s 拿到面条" % (name))

    print("开始吃面条 ... ")
    time.sleep(0.5)

    print("%s 放下面条" % (name))    
    print("%s 放下筷子" % (name))
    mylock.release()

if __name__ == "__main__":
    name_lst1 = ["周永玲","张龙"]
    name_lst2 = ["李诗韵","黄乐锡"]

    for name in name_lst1:
        Thread(target=eat1,args=(name,)).start()
        
    for name in name_lst2:
        Thread(target=eat2,args=(name,)).start()

 

# ### 线程之间的数据隔离

　　首先我们要明白线程与线程之间是共享数据的，那么线程之间是如何实现数据之间的隔离的呢？

　　from threading import Thread,local,currentThread

　　使用threading模块下面的local方法可以实现数据隔离

　　

# 创建对象
loc = local()
# 为当前loc对象的val属性赋值
loc.val = "main_thread 目前下载的进度 57%"
print(loc , loc.val)

# func2 负责打印数据
def func2():
    # loc.val,currentThread().name 获取当前线程名 MainThread
    print("%s 该数据归属于 %s" % (loc.val,currentThread().name))
    
# 另外的一个子线程下载任务
def func1(speed):
    loc.val = speed
    func2()

# func2()

t1 = Thread(target=func1,args=("下载进度%9",))
t1.setName("子线程1")
t1.start()

t2 = Thread(target=func1,args=("下载进度%19",))
t2.setName("子线程2")
t2.start()

由上面我们可以看出线程与线程之间数据彼此隔离是通过local.val这个属性来存放的，间接造成的数据隔离

# ### 事件 Event 与传统进出也能够吃的使用方法是一样的，都是判定is_set（）当前的状态，默认是False ,wait 是等待，set是把默认的false设置为True  ,当处于True 时不阻塞，反之阻塞

    e = Event()
    
    # wait 动态添加阻塞
    # is_set() 判断当前的状态属性值(True False) 默认False
    # clear()  将内部的成员属性值改成False
    # set()    将内部的成员属性值改成True    
    
e = Event()
print(e.is_set())
e.set()  # 将属性值改成True
e.clear()# 将属性值改成False
e.wait() # True 放行 False 阻塞
print(111)

def func(e):
time.sleep(5)
e.set()

e = Event()
t = Thread(target=func,args=(e,))
t.start()

print(e.is_set())

# 3 => 最多阻塞3秒钟时间
e.wait(1)
print("主线程执行结束")

(2) 模拟链接数据库

 

def check(e):
    print("开始检测用户连接的合法性")
    time.sleep(random.randrange(1,6))
    e.set()

# 尝试连接数据库
def connect(e):
    sign = False
    # 尝试连接三次
    for i in range(3):
        e.wait(1)
        if e.is_set():
            sign = True
            print("数据库连接成功")
            break
        else:
            print("尝试连接数据%s次失败" % (i+1))
            
    if sign == False:
        # 主动抛出异常
        raise TimeoutError

e = Event()
# 执行check 任务
Thread(target=check,args=(e,)).start()

# 执行connect 任务
Thread(target=connect,args=(e,)).start()

 线程队列事件 # ### 队列

线程是共享一个数据的，队列对于线程来说有点多余，而对于进程来说就不一样了，进程与进程之间的数据彼此隔离，下面我来介绍下线程之间的队列是如何实现的

　　

　　put 往队列里放数据
　　get 从队列里取数据
　　put_nowait 往队列中放的值如果超过了队列长度,直接报错
　　get_nowait 从队列中取值,如果队列中没有值,直接报错

# (1) Queue 先进先出
q = Queue()
q.put(1)
q.put(2)
print(q.get())
print(q.get())
# 队列中没有值,会阻塞
# q.get()
# 取不到值, 直接报错,可以使用try .. except ..来抑制错误
# q.get_nowait()

q2 = Queue(3)
q2.put(1)
q2.put(2)
q2.put(3)
# 超出队列元素个数,直接阻塞
# q2.put(4)
# 超出队列元素个数,直接报错
# q2.put_nowait(4)

(2) lifoQueue 后进先出,先进后出, 模拟栈这个数据结构特点

　　from queue import LifoQueue
　　lq = LifoQueue()
　　lq.put(10)
　　lq.put(11)
　　print(lq.get())#11

 (3) PriorityQueue 按照优先级顺序来进行排序 (默认从小到大排序)

from queue import PriorityQueue
pq = PriorityQueue()

# 1 (数字,字符串) 按照数字排序
pq.put( (18 , "李诗韵") )
pq.put( (81 , "周永玲") )
pq.put( (60 , "张龙") )
pq.put( (36 , "张晓东") )

print(pq.get())
print(pq.get())
print(pq.get())
print(pq.get())

# 2 (字符串,数字) 按照ascii编码排序
pq.put( ("zhangsan",90))
pq.put( ("lisi",91))
pq.put( ("wangwu",13))
pq.put( ("zhaoliu",17))

print(pq.get())
print(pq.get())
print(pq.get())
print(pq.get())

# 3 只能插入同一类型,不能混合插入
# pq.put(18)
# pq.put("nihao")

# ### 进程池 , 线程池 (改良版)

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import os,time
def func(i):
    print("进程号:",os.getpid())
    print(i)
    time.sleep(2)
    print("当前进程执行结束 end ...")
    return 123

　　(1)ProcessPoolExecutor 进程池基本使用

　　"""新版进程池默认主进程在等待所有子进程执行结束之后,在终止程序,释放资源"""

　　

if __name__ == "__main__":
    # (1) 创建进程池对象 ProcessPoolExecutor <==> Pool
    """默认参数 : os.cpu_count() 是cpu逻辑核心数"""
    p = ProcessPoolExecutor()
    # print(os.cpu_count())
    # (2) 提交异步任务 submit <==> apply_async
    res = p.submit(func,66)
    print(res)
    # (3) 获取返回值 result <==> get 内部也存在阻塞
    res = res.result()
    print(res)
    # (4) shutdown <==> close + join
    p.shutdown()
    print("主进程执行结束 ... ")

(2) ThreadPoolExecutor 线程池基本用法

from threading import currentThread as cthread
def func(i):
    print("thread",i,cthread().ident)
    # time.sleep(0.1)
    print("thread %s end" % (i))
    
# 同一时间最多允许并行3个线程
tp = ThreadPoolExecutor()
for i in range(20):
    tp.submit(func,i)

tp.shutdown()
print("主线程执行结束")    

　　(3) 获取线程池的返回值

def func(i):
    print("thread" , i , cthread().ident)
    print("thread %s end ..."%(i))
    # time.sleep(0.1) 为了触发更多的线程
    return cthread().ident
    
tp = ThreadPoolExecutor(5)
lst = []
setvar = set()
for i in range(10):
    res = tp.submit(func,i)
    lst.append(res)
    
for i in lst:
    setvar.add(i.result())
    
tp.shutdown()
print(setvar)
print("主线程执行结束 ...")

(4) map 返回迭代器
def func(i):
    print("thread",i,cthread().ident)
    time.sleep(0.1)
    print("thread %s end" % (i))
    return "*" * i

tp = ThreadPoolExecutor(5)
it = tp.map(func,range(20))
for i in it:
    print(i)

from collections import Iterator
print(isinstance(it,Iterator))

 什么是回调函数？？

　　回调函数:
　　把函数当成参数传递给另外一个函数
　　在当前函数执行完毕之后,调用一下传递进来的函数,该函数是回调函数

即当前函数执行的过程中执行传入进来的参数，而传进来的参数本身也是一个函数那么我当前函数执行的过程称为回调函数

而线程池中也存在回调函数的概念，

(1)add_done_callback 在获取当前线程的返回值的时候,可以异步并发,加快速度
(2)回调函数由谁执行:由执行任务的当前子线程,调用回调函数

如下

from concurrent.futures import ThreadPoolExecutor
from threading import currentThread as cthread
import time,os            

lst = []
def func(i):
    print("thread",i,cthread().ident)
    time.sleep(0.1)
    print("thread %s end" % (i))
    return "*" * i

# 回调函数
def call_back(obj):
    print("<==start==>")
    print("call_back:",cthread().ident)
    print(obj)
    print(obj.result())
    print("<==end==>")

tp = ThreadPoolExecutor(5)
for i in range(1,11):
    obj = tp.submit(func,i)
    # 使用回调函数,异步并发
    obj.add_done_callback(call_back)
    # print(obj)
    # lst.append(obj)

class MyClass():
def add_done_callback(self,func):
# code1 code2 code3....
"""
self <==> obj
func <==> call_back
"""
# 最后func
func(self)

def call_back(args):
# args <==> obj
print("call_back:",cthread().ident)
print(args)
print(args.result())
obj = MyClass()
obj.add_done_callback(call_back)

 (2) 进程池,它的回调函数 由主进程完成

　　(1)add_done_callback 在获取当前进程的返回值的时候,可以异步并发,加快速度
　　(2)回调函数由谁执行:都由主进程来完成

from concurrent.futures import ProcessPoolExecutor
def func(i):
    print("Process",i,os.getpid())
    time.sleep(0.1)
    print("Process %s end" % (i))
    return "*" * i

def call_back(obj):
    print("call_back:",os.getpid())
    print(obj.result())

if __name__ == "__main__":
    p = ProcessPoolExecutor(5)
    for i in range(1,11):
        obj = p.submit(func,i)
        # print(obj.result())
        obj.add_done_callback(call_back)

    print("主进程执行结束.." , os.getpid())