并发编程 - 线程 - 1.互斥锁/2.GIL解释器锁/3.死锁与递归锁/4.信号量/5.Event事件/6.定时器

Posted 2020-10-29 Alice的小屋

1.互斥锁：
    原理：将并行变成串行
    精髓：局部串行，只针对共享数据修改
          保护不同的数据就应该用不用的锁

 1 from threading import Thread, Lock
 2 import time
 3 
 4 n = 100
 5 
 6 def task():
 7     global n
 8     mutex.acquire()  # 效率低了 但是数据安全了
 9     temp = n
10     time.sleep(0.1)  # 100个线程 都拿到了100  所以就是 100个线程100-1
11     n = temp - 1
12     mutex.release()
13 
14 
15 if __name__ == \'__main__\':
16     mutex = Lock()
17     t_l = []
18     for i in range(100):
19         t = Thread(target=task)
20         t_l.append(t)
21         t.start()
22 
23     for t in t_l:
24         t.join()
25 
26     print(\'主\', n)
27 """
28 主 99   原因： 100个线程 都拿到了100  所以就是 100个线程100-1  数据不安全 效率高但是不安全
29           要将并行改为串行 
30 """
31 """
32 主 0    原因：效率低了 但是数据安全了
33 """

2.GIL: global interpreter lock
    python3 test.py
    ps aux | grep test     # linux
    tasklist | findstr python   # windows  python.exe

运行python 会有几步：
    1.会有一个进程，进程内存空间 python解释器的代码先加载到内存空间
    2.test.py 内容加载到内存
    3.解释执行；代码交给了python解释器
        线程干活指向了python代码 python代码当作参数传给了解释器
        线程拿到解释器的代码，拿着python代码当作参数，执行
        垃圾回收线程运行解释器的代码
        垃圾回收线程和某一个线程冲突了，数据不安全，
   

 

    开多个进程，GIL就没影响了， cpython解释器垃圾回收线程定期启动一个
    GIL:互斥锁，保证数据的安全 对CPython解释器，同一时间只有一个线程运行
        GIL.acquire()  这样垃圾线程和线程就不会冲突了，这样回收机制就变得安全了
        GIL.release()
    python解释器，多线程有GIL存在，保证了一个进程下面多个线程的执行是一个一个执行的

   有GIL与自动的锁的工作原理：
       

 

   总结：
        1.GIL 一个进程内的多个线程同一时间只能运行一个线程，垃圾回收线程是安全的
        2.针对不同的数据，就应该加不同的锁，解释器级别的GIL锁，只能保护解释器级别的数据，
          不能保护自己的数据，针对自己的共享数据还要加锁；

          线程首先抢的是；GIL锁，之后才是mutex
   官网：
      结论：在Cpython解释器中，同一个进程下开启的多线程，同一时刻只能有一个线程执行，无法利用多核优势

   GIL的存在：同一时刻，只能有一个线程在运行
        多核，多进程，但进程开销大，多线程，又不能用多核 ?

   cpu干计算的，多个cpu
        1.如果是干计算的操作，多核省时间
        2.如果干IO阻塞型操作，多核没用

   程序运行：都会干计算和IO操作

   四个任务：
        1.1个核：开多线程  ，因为多进程能用上多核
        2.多核:
            计算密集型：用多进程，用多核，eg:金融行业的，计算比较多,虽然多进程开销大，但多核，保证了计算快
            IO密集型：用多线程，同一时间只能用一个核，1个核一个进程，多线程就在一个核上来回切和四个核来回切是一样的

   现在写的软件：
      网络打交道，网络的IO
      IO密集型，用多线程

 1 """
 2 计算密集型应该用： 多进程 效率高  
 3 """
 4 from multiprocessing import Process
 5 from threading import Thread
 6 import os,time
 7 
 8 def work():
 9     res=0
10     for i in range(100000000):
11         res*=i
12 
13 
14 if __name__ == \'__main__\':
15     l=[]
16     print(os.cpu_count()) #本机为8核
17     start=time.time()
18     for i in range(8):
19         # p=Process(target=work) #耗时8s多
20         p=Thread(target=work) #耗时37s多
21         l.append(p)
22         p.start()
23     for p in l:
24         p.join()
25     stop=time.time()
26     print(\'run time is %s\' %(stop-start))
27 
28 """
29 IO密集型：多线程 效率高
30 """
31 from multiprocessing import Process
32 from threading import Thread
33 import threading
34 import os,time
35 def work():
36     time.sleep(2)
37     print(\'===>\')
38 
39 if __name__ == \'__main__\':
40     l=[]
41     print(os.cpu_count()) #本机为8核
42     start=time.time()
43     for i in range(400):
44         # p=Process(target=work) #耗时8s多,大部分时间耗费在创建进程上
45         p=Thread(target=work) #耗时2s多
46         l.append(p)
47         p.start()
48     for p in l:
49         p.join()
50     stop=time.time()
51     print(\'run time is %s\' %(stop-start))

3.死锁：
    你拿着我的锁，我拿着你的锁

互斥锁：Lock()
    互斥锁只能acquire一次

递归锁:RLock()
    可以连续acquire多次，每acquire一次计数器+1，
    只有计数为0时，才能被抢到acquire

 1 from threading import Thread,Lock
 2 import time
 3 
 4 mutexA=Lock()
 5 mutexB=Lock()
 6 
 7 class MyThread(Thread):
 8     def run(self):
 9         self.f1()
10         self.f2()
11 
12     def f1(self):
13         mutexA.acquire()
14         print(\'%s 拿到了A锁\' %self.name)
15 
16         mutexB.acquire()
17         print(\'%s 拿到了B锁\' %self.name)
18         mutexB.release()
19 
20         mutexA.release()
21 
22 
23     def f2(self):
24         mutexB.acquire()
25         print(\'%s 拿到了B锁\' % self.name)
26         time.sleep(0.1)
27 
28         mutexA.acquire()
29         print(\'%s 拿到了A锁\' % self.name)
30         mutexA.release()
31 
32         mutexB.release()
33 
34 if __name__ == \'__main__\':
35     for i in range(10):
36         t=MyThread()
37         t.start()
38 """
39 Thread-1 拿到了A锁  # 死锁了 卡住了
40 Thread-1 拿到了B锁
41 Thread-1 拿到了B锁
42 Thread-2 拿到了A锁
43 """

 1 # 互斥锁只能acquire一次
 2 # from threading import Thread,Lock
 3 #
 4 # mutexA=Lock()
 5 #
 6 # mutexA.acquire()
 7 # mutexA.release()
 8 
 9 # 递归锁:可以连续acquire多次，每acquire一次计数器+1，只有计数为0时，才能被抢到acquire
10 from threading import Thread,RLock
11 import time
12 
13 mutexB=mutexA=RLock()
14 
15 class MyThread(Thread):
16     def run(self):
17         self.f1()
18         self.f2()
19 
20     def f1(self):
21         mutexA.acquire()
22         print(\'%s 拿到了A锁\' %self.name)
23 
24         mutexB.acquire()
25         print(\'%s 拿到了B锁\' %self.name)
26         mutexB.release()
27 
28         mutexA.release()
29 
30 
31     def f2(self):
32         mutexB.acquire()
33         print(\'%s 拿到了B锁\' % self.name)
34         time.sleep(2)
35 
36         mutexA.acquire()
37         print(\'%s 拿到了A锁\' % self.name)
38         mutexA.release()
39 
40         mutexB.release()
41 
42 if __name__ == \'__main__\':
43     for i in range(10):
44         t=MyThread()
45         t.start()
46 """
47 Thread-1 拿到了A锁  # 解决了 死锁
48 Thread-1 拿到了B锁
49 Thread-1 拿到了B锁
50 Thread-1 拿到了A锁
51 Thread-2 拿到了A锁
52 Thread-2 拿到了B锁
53 Thread-2 拿到了B锁
54 Thread-2 拿到了A锁
55 Thread-4 拿到了A锁
56 Thread-4 拿到了B锁
57 Thread-5 拿到了A锁
58 Thread-5 拿到了B锁
59 Thread-5 拿到了B锁
60 Thread-5 拿到了A锁
61 Thread-7 拿到了A锁
62 Thread-7 拿到了B锁
63 Thread-7 拿到了B锁
64 Thread-7 拿到了A锁
65 Thread-9 拿到了A锁
66 Thread-9 拿到了B锁
67 Thread-9 拿到了B锁
68 Thread-9 拿到了A锁
69 Thread-3 拿到了A锁
70 Thread-3 拿到了B锁
71 Thread-3 拿到了B锁
72 Thread-3 拿到了A锁
73 Thread-6 拿到了A锁
74 Thread-6 拿到了B锁
75 Thread-6 拿到了B锁
76 Thread-6 拿到了A锁
77 Thread-10 拿到了A锁
78 Thread-10 拿到了B锁
79 Thread-10 拿到了B锁
80 Thread-10 拿到了A锁
81 Thread-8 拿到了A锁
82 Thread-8 拿到了B锁
83 Thread-8 拿到了B锁
84 Thread-8 拿到了A锁
85 Thread-4 拿到了B锁
86 Thread-4 拿到了A锁
87 """

4.信号量
　　信号量也是一把锁，可以指定信号量为5，对比互斥锁同一时间只能有一个任务抢到锁去执行，
   　　 信号量同一时间可以有5个任务拿到锁去执行
    　　信号量：同一时间有多个线程在进行

 1 from threading import Thread,Semaphore,currentThread
 2 import time,random
 3 
 4 sm=Semaphore(1)
 5 
 6 def task():
 7     # sm.acquire()
 8     # print(\'%s in\' %currentThread().getName())
 9     # sm.release()
10     with sm:  # 类似于sm.acquire() # 同一时间可以来3个人，1个人，或者2个人
11         print(\'%s in\' %currentThread().getName())
12         time.sleep(random.randint(1,3))
13 
14 
15 if __name__ == \'__main__\':
16     for i in range(10):
17         t=Thread(target=task)
18         t.start()
19 """
20 Thread-1 in
21 Thread-2 in
22 Thread-3 in
23 
24 Thread-4 in
25 
26 
27 Thread-6 in
28 Thread-5 in
29 Thread-7 in
30 
31 
32 Thread-8 in
33 Thread-9 in
34 
35 Thread-10 in 
36 """

5.Event:
   多个线程之间同步的，一个线程告诉另一些线程可以做其他的活了
   event.wait()
   event.wait(2)
   event.set()
   event.is_set()
   event.clear()

 1 from threading import Thread,Event
 2 import time
 3 
 4 event=Event()
 5 # event.wait()   # 等 ...直到 set
 6 # event.set()
 7 
 8 
 9 def student(name):
10     print(\'学生%s 正在听课\' %name)
11     # event.wait()  # 学生要等7秒 才能下课
12     event.wait(2)   # 学生等2秒 直接下课了
13 
14     print(\'学生%s 课间活动\' %name)
15 
16 
17 def teacher(name):
18     print(\'老师%s 正在授课\' %name)
19     time.sleep(7)
20     event.set()
21 
22 
23 if __name__ == \'__main__\':
24     stu1=Thread(target=student,args=(\'alex\',))
25     stu2=Thread(target=student,args=(\'wxx\',))
26     stu3=Thread(target=student,args=(\'yxx\',))
27     t1=Thread(target=teacher,args=(\'egon\',))
28 
29     stu1.start()
30     stu2.start()
31     stu3.start()
32     t1.start()
33 
34 
35 # ------------------
36 # 设置链接的超时时间
37 from threading import Thread,Event,currentThread
38 import time
39 
40 event=Event()
41 
42 def conn():
43     # print(\'%s is connecting\'%currentThread().getName())
44     # event.wait()
45     # print(\'%s is connected\'%currentThread().getName())
46 
47     n=0
48     while not event.is_set():
49         if n == 3:
50             print(\'%s try too many times\' %currentThread().getName())
51             return
52         print(\'%s try %s\' %(currentThread().getName(),n))
53         event.wait(0.5)
54         n+=1
55 
56     print(\'%s is connected\' %currentThread().getName())
57 
58 
59 def check():
60     print(\'%s is checking\' %currentThread().getName())
61     time.sleep(5)
62     event.set()
63 
64 
65 if __name__ == \'__main__\':
66     for i in range(3):
67         t=Thread(target=conn)
68         t.start()
69     t=Thread(target=check)
70     t.start()
71 """
72 Thread-1 try 0
73 Thread-2 try 0
74 Thread-3 try 0
75 Thread-4 is checking
76 Thread-3 try 1
77 Thread-2 try 1
78 Thread-1 try 1
79 Thread-3 try 2
80 Thread-1 try 2
81 Thread-2 try 2
82 Thread-3 try too many times
83 Thread-2 try too many times
84 Thread-1 try too many times
85 """

6.定时器：Timer
　　t=Timer(5,task,args=(\'egon\',))
　　t.start()
　　t.cancel()

 1 from threading import Timer
 2 
 3 def task(name):
 4     print(\'hello %s\' %name)
 5 
 6 t=Timer(5,task,args=(\'egon\',))  # 就是起了一个线程
 7 t.start()
 8 
 9 # ----------------------
10 from threading import Timer
11 import random
12 
13 class Code:
14     def __init__(self):
15         self.make_cache()
16 
17     def make_cache(self,interval=10):
18         self.cache=self.make_code()
19         print(self.cache)
20         self.t=Timer(interval,self.make_cache)
21         self.t.start()
22 
23     def make_code(self,n=4):
24         res=\'\'
25         for i in range(n):
26             s1=str(random.randint(0,9))
27             s2=chr(random.randint(65,90))
28             res+=random.choice([s1,s2])
29         return res
30 
31     def check(self):
32         while True:
33             code=input(\'请输入你的验证码>>: \').strip()
34             if code.upper() == self.cache:
35                 print(\'验证码输入正确\')
36                 self.t.cancel()
37                 break
38 
39 obj=Code()
40 obj.check()