多线程的理解

Posted 2020-09-14 _潜行者

 

• python的threading.Thread类有一个run方法，用于定义线程的功能函数，可以在自己的线程类中覆盖该方法。而创建自己的线程实例后，通过Thread类的start方法，可以启动该线程，交给python虚拟机进行调度，当该线程获得执行的机会时，就会调用run方法执行线程。
• 每个线程一定会有一个名字，尽管上面的例子中没有指定线程对象的name，但是python会自动为线程指定一个名字。
• 当线程的run()方法结束时该线程完成。
• 在一个进程内的所有线程共享全局变量，能够在不适用其他方式的前提下完成多线程之间的数据共享（这点要比多进程要好）
• 缺点就是，线程是对全局变量随意遂改可能造成多线程之间对全局变量的混乱（即线程非安全）

和进程之间的区别：

• 进程是系统进行资源分配和调度的一个独立单位.

• 线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程自己基本上不拥有系统资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈),但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源.

• 在多线程开发中，全局变量是多个线程都共享的数据，而局部变量等是各自线程的，是非共享的

• 区别

  • 一个程序至少有一个进程,一个进程至少有一个线程.

  • 线程的划分尺度小于进程(资源比进程少)，使得多线程程序的并发性高。

  • 进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大地提高了程序的运行效率

  • 线线程不能够独立执行，必须依存在进程中

  优缺点

  线程和进程在使用上各有优缺点：线程执行开销小，但不利于资源的管理和保护；而进程正相反。

 

使用互斥锁实现线程同步

from threading import Thread,Lock
from time import sleep

class Task1(Thread):
    def run(self):
        while True:
            if lock1.acquire():
                print("------Task 1 -----")
                sleep(0.5)
                lock2.release()

class Task2(Thread):
    def run(self):
        while True:
            if lock2.acquire():
                print("------Task 2 -----")
                sleep(0.5)
                lock3.release()

class Task3(Thread):
    def run(self):
        while True:
            if lock3.acquire():
                print("------Task 3 -----")
                sleep(0.5)
                lock1.release()

#使用Lock创建出的锁默认没有“锁上”
lock1 = Lock()
#创建另外一把锁，并且“锁上”
lock2 = Lock()
lock2.acquire()
#创建另外一把锁，并且“锁上”
lock3 = Lock()
lock3.acquire()

t1 = Task1()
t2 = Task2()
t3 = Task3()

t1.start()
t2.start()
t3.start()

　　运行结果：

------Task 1 -----
------Task 2 -----
------Task 3 -----
------Task 1 -----
------Task 2 -----
------Task 3 -----
------Task 1 -----
------Task 2 -----
------Task 3 -----
------Task 1 -----
------Task 2 -----
------Task 3 -----
------Task 1 -----
------Task 2 -----
------Task 3 -----
...省略...

重要方法：

t.setDaemon(True)#默认是False,True表示主线程执行到程序结尾不管子线程是否执行完，都结束程序
 """A boolean value indicating whether this thread is a daemon thread.

 This must be set before start() is called, otherwise RuntimeError is
raised. Its initial value is inherited from the creating thread; the
main thread is not a daemon thread and therefore all threads created in
the main thread default to daemon = False.

The entire Python program exits when no alive non-daemon threads are
 left.

"""
同样的进程中也有类似方法p.Daemon = True

主线程等待子线程执行：(同样适用于进程通信，进程join方法没有参数)

t.join() #等待子线程执行完然后主线程接着执行，变态用法，这样子线程的并发就没有意义

t.join(2)#与上面类似，不过最多等2秒

　　

进程间数据不共享（与多线程不同点）：

 

from multiprocessing import Pool, Process
import os, time

l = []

def test(temp):
    time.sleep(1)
    l.append(temp)

    print("{}say hai".format(os.getpid()), l)

if __name__ == \'__main__\':
    for i in range(10):
        p = Process(target=test, args=(i, ))
        p.start()
        p.join()#1秒执行一个（打印一次），注销此行，那么各个进程间不在等待，1秒后几乎同时输出

 

　　打印

/usr/local/Cellar/python3/3.6.0_1/Frameworks/Python.framework/Versions/3.6/bin/python3.6 /Users/starhub/PycharmProjects/多线程/1.py
38000say hai [0]
38001say hai [1]
38002say hai [2]
38003say hai [3]
38004say hai [4]
38005say hai [5]
38006say hai [6]
38008say hai [7]
38010say hai [8]
38011say hai [9]

　　如果是线程那么打印结果将是下面这样：

/usr/local/Cellar/python3/3.6.0_1/Frameworks/Python.framework/Versions/3.6/bin/python3.6 /Users/starhub/PycharmProjects/多线程/1.py
38322say hai [1]
38322say hai [1, 0]
38322say hai [1, 0, 2]
38322say hai [1, 0, 2, 3]
38322say hai [1, 0, 2, 3, 5]
38322say hai [1, 0, 2, 3, 5, 7]
38322say hai [1, 0, 2, 3, 5, 7, 9]
38322say hai [1, 0, 2, 3, 5, 7, 9, 4]
38322say hai [1, 0, 2, 3, 5, 7, 9, 4, 8]
38322say hai [1, 0, 2, 3, 5, 7, 9, 4, 8, 6]

　　如果想进程间数据共享，那么可以使用特殊的数据结构（Array，Manager.dict,）

 

event

 

执行结果：

 

 

使用threading.local()方法

 

 1 import threading
 2 
 3 # 创建全局ThreadLocal对象:
 4 local_school = threading.local()
 5 
 6 def process_student():
 7     # 获取当前线程关联的student:
 8     std = local_school.student
 9     print(\'Hello, %s (in %s)\' % (std, threading.current_thread().name))
10 
11 def process_thread(name):
12     # 绑定ThreadLocal的student:
13     local_school.student = name
14     process_student()
15 
16 t1 = threading.Thread(target= process_thread, args=(\'dongGe\',), name=\'Thread-A\')
17 t2 = threading.Thread(target= process_thread, args=(\'老王\',), name=\'Thread-B\')
18 t1.start()
19 t2.start()
20 t1.join()
21 t2.join()
22 执行结果：
23 
24 Hello, dongGe (in Thread-A)
25 Hello, 老王 (in Thread-B)

 

小结：

全局变量local_school就是一个ThreadLocal对象，每个Thread对它都可以读写student属性，但互不影响。你可以把local_school看成全局变量，但每个属性如local_school.student都是线程的局部变量，可以任意读写而互不干扰，也不用管理锁的问题，ThreadLocal内部会处理。

可以理解为全局变量local_school是一个dict，不但可以用local_school.student，还可以绑定其他变量，如local_school.teacher等等。

ThreadLocal最常用的地方就是为每个线程绑定一个数据库连接，HTTP请求，用户身份信息等，这样一个线程的所有调用到的处理函数都可以非常方便地访问这些资源。

一个ThreadLocal变量虽然是全局变量，但每个线程都只能读写自己线程的独立副本，互不干扰。ThreadLocal解决了参数在一个线程中各个函数之间互相传递的问题

 

队列（生产者与消费者）

 进程池实现同步和异步：

 1 from multiprocessing import Pool
 2 import time
 3 import os
 4 
 5 def test():
 6     print("---进程池中的进程---pid=%d,ppid=%d--"%(os.getpid(),os.getppid()))
 7     for i in range(3):
 8         print("----%d---"%i)
 9         time.sleep(1)
10     return "hahah"
11 
12 def test2(args):
13     print("---callback func--pid=%d"%os.getpid())
14     print("---callback func--args=%s"%args)
15 
16 pool = Pool(3)
17 pool.apply(func=test)#同步进程
18 \'\'\'
19 ---进程池中的进程---pid=36031,ppid=36030--
20 ----0---
21 ----1---
22 ----2---
23 ----主进程-pid=36030----
24 \'\'\'
25 pool.apply_async(func=test,callback=test2)
26 time.sleep(5)#结果如下如果没有延迟，那么执行结果apply()，主进程没有等待异步进程
27 \'\'\'
28 ---进程池中的进程---pid=36180,ppid=36179--
29 ----0---
30 ----1---
31 ----2---
32 ---进程池中的进程---pid=36181,ppid=36179--
33 ----0---
34 ----1---
35 ----2---
36 ---callback func--pid=36179
37 ---callback func--args=hahah
38 ----主进程-pid=36179----
39 \'\'\'
40 
41 print("----主进程-pid=%d----"%os.getpid())

进程池异步 apply_async()方法阐明Pool(5)参数意义

 1 from multiprocessing import Pool, Process
 2 # from threading import Thread
 3 import os, time
 4 
 5 
 6 def test(temp):
 7     time.sleep(1)
 8 
 9     print("{}say hai".format(os.getpid()), temp)
10 
11 if __name__ == \'__main__\':
12     p = Pool(4)
13     for i in range(40):
14         p.apply_async(func= test,args=(i, ))
15 
16     time.sleep(2)
17     print(\'主进程执行完毕\')

打印结果：

#/usr/local/Cellar/python3/3.6.0_1/Frameworks/Python.framework/Versions/3.6/bin/python3.6 /Users/starhub/PycharmProjects/多线程/1.py
39798say hai 0
39799say hai 1
39800say hai 2
39801say hai 3
39798say hai 4
39799say hai 5#下面的6，7出现的时间有延迟，证明进程最大并发数就是Pool（5）初始化参数5
39800say hai 6
39801say hai 7
主进程执行完毕

 1 from multiprocessing import Pool, Process
 2 # from threading import Thread
 3 import os, time
 4 
 5 
 6 def test(temp):
 7     time.sleep(1)
 8 
 9     print("{}say hai".format(os.getpid()), temp)
10 
11 if __name__ == \'__main__\':
12     p = Pool(4)
13     for i in range(40):
14         p.apply_async(func= test,args=(i, ))
15 
16     p.close()#关闭Pool，使其不再接受新的任务
17     p.join()#线程join必须配合close使用，等待线程池所有进程执行完毕。完全失去了异步进程的意义(主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；)
18     print(\'主进程执行完毕\')

执行结果是隔一秒打印5行，即执行5个

40138say hai 1
40137say hai 0
40139say hai 2
40140say hai 3
40138say hai 4
40137say hai 5
40139say hai 6
40140say hai 7
40137say hai 8
40138say hai 9
40139say hai 10
40140say hai 11
40137say hai 12
40138say hai 13
40139say hai 14
40140say hai 15
40138say hai 16
40137say hai 17
40139say hai 18
40140say hai 19
40138say hai 20
40137say hai 21
40139say hai 22
40140say hai 23
40138say hai 24
40137say hai 25
40139say hai 26
40140say hai 27
40138say hai 28
40137say hai 29
40139say hai 30
40140say hai 31
40137say hai 32
40139say hai 34
40138say hai 33
40140say hai 35
40137say hai 36
40139say hai 37
40138say hai 38
40140say hai 39
主进程执行完毕