多线程(Threading)和多进程(Multiprocessing)_threading和multiprocessing

---

线程和进程是什么

进程

• 是程序的一次执行过程，是程序在执行过程中分配和管理资源的基本单位
• 每一个进程都有一个自己的地址空间`
• 至少有 5 种基本状态，它们是：初始态，执行态，等待状态，就绪状态，终止状态。
  
  线程
• 是CPU`调度和分派的基本单位
• 与同属一个进程的其他的线程共享进程所拥有的全部资源,可以访问隶属进程的资源。

线程与进程的区别联系：

1. 线程是进程的一部分。一个线程只能属于一个进程，而一个进程可以有多个线程，但至少有一个线程：线程是进程的一部分。
2. 根本区别：进程是操作系统资源分配的基本单位，而线程是任务调度和执行的基本单位
3. 所处环境：在操作系统中能同时运行多个进程（程序）；而在同一个进程（程序）中有多个线程同时执行（通过CPU调度，在每个时间片中只有一个线程执行）
4. 内存分配：系统为每个进程分配不同的内存空间；而对线程而言，除了CPU外，系统不会为线程分配内存（线程所使用的资源来自其所属进程的资源），线程组之间只能共享资源

对于单核的CPU来说，真的可以做到真正的多线程并发吗？
对于多核的CPU电脑来说，真正的多线程并发是没问题的。4核CPU表示同一个时间点上，可以真正的有4个进程并发执行。

单核的CPU表示只有一个大脑：
不能够做到真正的多线程并发，但是可以做到给人一种“多线程并发”的感觉。
对于单核的CPU来说，在某一个时间点上实际上只能处理一件事情，但是由于CPU的处理速度极快，多个线程之间频繁切换执行，给别人的感觉是：多个事情同时在做！！！
eg.
线程A：播放音乐；线程B：运行游戏
线程A和线程B频繁切换执行，人类会感觉音乐一直在播放，游戏一直在运行，给我们的感觉是同时并发的。（因为计算机的速度很快，我们人的眼睛很慢，所以才会感觉是多线程！）

---

进程间通信方式

https://www.cnblogs.com/zgq0/p/8780893.html

1. 无名管道（pipe）

• 半双工（Half Duplex）
• 只能用于具有亲缘关系的进程之间的通信（也是父子进程或者兄弟进程之间）
• 它可以看成是一种特殊的文件，对于它的读写也可以使用普通的read、write 等函数。但是它不是普通的文件，并不属于其他任何文件系统，并且只存在于内存中

2. FIFO 命名管道

• 可以在无关的进程之间交换数据
• FIFO有路径名与之相关联，它以一种特殊设备文件形式存在于文件系统中
• FIFO的通信方式类似于在进程中使用文件来传输数据，只不过FIFO类型文件同时具有管道的特性。在数据读出时，FIFO管道中同时清除数据，并且“先进先出”

3. 消息队列（Message queue）
   是消息的链接表，存放在内核中。一个消息队列由一个标识符（即队列ID）来标识。

• 消息队列是面向记录的，其中的消息具有特定的格式以及特定的优先级。
• 消息队列独立于发送与接收进程。进程终止时，消息队列及其内容并不会被删除。
• 消息队列可以实现消息的随机查询,消息不一定要以先进先出的次序读取,也可以按消息的类型读取。

4. 信号量（semaphore）
   它是一个计数器。信号量用于实现进程间的互斥与同步，而不是用于存储进程间通信数据。
5. 共享内存（Shared Memory）
   指两个或多个进程共享一个给定的存储区。信号量+共享内存通常结合在一起使用，信号量用来同步对共享内存的访问。

总结：
1.管道：速度慢，容量有限，只有父子进程能通讯
2.FIFO：任何进程间都能通讯，但速度慢
3.消息队列：容量受到系统限制，且要注意第一次读的时候，要考虑上一次没有读完数据的问题
4.信号量：不能传递复杂消息，只能用来同步
5.共享内存区：能够很容易控制容量，速度快，但要保持同步，比如一个进程在写的时候，另一个进程要注意读写的问题，相当于线程中的线程安全，当然，共享内存区同样可以用作线程间通讯，不过没这个必要，线程间本来就已经共享了同一进程内的一块内存

---

线程间通信方式

https://blog.csdn.net/qq_42473704/article/details/81942347

线程的生命周期

---

死锁

死锁产生的4个必要条件：
1、互斥：一个资源同一时刻只允许一个线程进行访问。
2、占有未释放：一个线程占有资源，且没有释放资源。
3、不可抢占：一个已经占有资源的线程无法抢占到其他线程拥有的资源。
4、循环等待：两个或者两个以上的线程，本身拥有资源，不释放资源，并且同时尝试获得其他线程所持有的资源，这种资源的申请关系形成一个闭环的链条。

解决办法：

1. 死锁检测与恢复的方案是引入看门狗计数器。当线程正常 运行的时候会每隔一段时间重置计数器，在没有发生死锁的情况下，一切都正常进行。一旦发生死锁，由于无法重置计数器导致定时器 超时，这时程序会通过重启自身恢复到正常状态。
2. 在进程获取锁的时候会严格按照对象id升序排列获取。避免死锁的主要思想是，单纯地按照对象id递增的顺序加锁不会产生循环依赖，而循环依赖是 死锁的一个必要条件，从而避免程序进入死锁状态。

---

Python多线程 Threading

什么是多线程

单线程：
在程序当中，只有一个箭头指向代码从头到尾执行。

import time
def sorry():
    print("Sorry")
    time.sleep(1)

if __name__ == '__main__':
    for i in range(5):
        sorry()
1
2
3
4
5
6
7
8

依次打印五个sorry，每个间隔1秒

多线程：
在一个脚本当中，一次同时运行多个程序。

import time
import threading
def sorry():
    print("Sorry")
    time.sleep(1)

if __name__ == '__main__':
    for i in range(5):
    	# Thread的参数为键 - 值对，target = 调用他的函数
        t = threading.Thread(target = sorry)	
        # 要用start启动线程
        t.start()
1
2
3
4
5
6
7
8
9
10
11
12

这是五个线程同时启动，五个“Sorry”同时打印出来。-- > 并发

基本方法函数

添加线程：

t = threading.Thread(target = thread_job)
t.start()
1
2

有多少个激活了的线程：

threading.active_count()
1

线程详情：

threading.enumerate()
threading.current_thread()
1
2

join（）

使得join以后的语句，必须要等到t线程的所有语句执行完以后才执行

import threading
import time
def thread_job():
    print("T1 start")
    for i in range(10):
        time.sleep(1)
    print("T1 finished")

def main():
    t = threading.Thread(target = thread_job, name = 'T1')
    t.start()
    t.join()
    print("all done")
if __name__ == '__main__':
    main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

T1会总计sleep才会输出“ T1 finished”，应该会先print “all done”出来。
但是由于t.join（），所以后面的语句会等T1执行完才执行。

T1 start
T1 finished
all done
1
2
3

Queue

因为thread是没有返回值的，不能像函数一样返回。所以要用queue来储存结果，用于输出。

import threading
import time
from queue import Queue

def job(l, q):
    for i in range(len(l)):
        l[i] = l[i] ** 2
    q.put(l)	# 储存结果到queue

def multithreading(data):
    q = Queue()	# 储存结果
    threads = []	# 储存所有线程
    for i in range(4):
        t = threading.Thread(target = job, args = (data[i], q))	# 目标函数所接收的参数
        t.start()
        threads.append(t)
    for thread in threads:
        thread.join()
    res = []
    for _ in range(4):
        res.append(q.get())	# 得到结果
    print(res)

if __name__ == '__main__':
    data = [[1, 2, 3], [3, 4, 5], [4, 4, 4], [5, 5, 5]]
    multithreading(data)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

继承使用线程

要重写父类的run方法。

import time
import threading
class MyThread(threading.Thread):   # 括号里写要继承的父类
    # 重写父类的run方法
    def run(self):  
        for i in range(5):
            print(f"I am {self.name} {str(i)}")

if __name__ == '__main__':
    t = MyThread()	# 实例化线程的类
    t.start()
1
2
3
4
5
6
7
8
9
10
11

五个线程同时打印：

I am Thread-1 0
I am Thread-1 1
I am Thread-1 2
I am Thread-1 3
I am Thread-1 4
1
2
3
4
5

1. 每个线程一定会有一个名字，self.name --> Thread - 1
2. 线程run（）方法结束，该线程结束
3. 无法控制线程的调度，因为操作系统说了算。可以通过别的方式来影响线程的调度：互斥、死锁、条件变量。
4. 线程的几种状态：新建 – 就绪 – 等待（阻塞）-- 运行态 – 死亡

同步

问题：
t1 t2两个线程，对一个全局变量num = 0进行增1运算，都对num修改十次
num永远为1，不会变成10.

两个不同的线程共享全局的资源，共享全局变量 --> 争夺！

使用多线程的时候，尽量避免全局变量，否则一定他要加锁lock（）

什么是同步：
协同步调，按照预定的先后顺序进行运行（如：单线程）。比如：你说完，我再说

锁实现同步

import time
import threading
num = 0 # 全局变量
# 创建一把锁
mutex = threading.Lock()    # 锁默认是开的
class MyThread(threading.Thread):
    def run(self):
        global num  # 声明要对全局变量进行修改
        mutexFlag = mutex.acquire()
        print(f"线程{self.name}的锁状态为{mutexFlag}")

        # 判断时候上锁成功
        if mutexFlag:
            num += 1
            time.sleep(1)
            print(f"{self.name} set num to {str(num)}")
            # 如果上锁以后不解开，就会“ 死锁 ”，程序卡在这里，不结束也不死亡
            # 解锁
            mutex.release()
def test():
    for i in range(5):
        t = MyThread()
        t.start()

if __name__ == '__main__':
    test()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

线程Thread-1的锁状态为True
Thread-1 set num to 1
线程Thread-2的锁状态为True
Thread-2 set num to 2
线程Thread-3的锁状态为True
Thread-3 set num to 3
线程Thread-4的锁状态为True
Thread-4 set num to 4
线程Thread-5的锁状态为True
Thread-5 set num to 5
1
2
3
4
5
6
7
8
9
10

锁：
好处：确保某段关键代码只能由一个线程从头到尾完整的执行
坏处：阻止了多线程并发运行，效率大大降低。由于可以存在多个锁，不同线程持有不同的锁，并且试图获取对方的锁，就可能会出现死锁。

---

GIL锁

GIL（Global Interpreter Lock）:

• 全局的互斥锁（mutex）
• 防止多线程同时操作一段字节码，为了Python解释器中原子操作的线程安全。

Python多线程不一定效率提升，因为python不是把任务平均分给每个线程同时做一个任务。实际上是靠GIL锁进行多线程，每次锁住一个线程进行任务，然后不断切换进程达成多线程的效果。python还是只能让一个线程在一个时间进行运算。

为了解决多线程之间数据完整性和状态同步的问题，设计为在任意时刻只有一个线程在解释器中运行。而当执行多线程程序时，由GIL来控制同一时刻只有一个线程能够运行。即Python中的多线程是表面多线程，也可以理解为fake多线程，不是真正的多线程。依靠解释器的分时复用，多个线程的代码，轮流被解释器执行。

普通解释：
并发：交替做不同事情的能力
并行：同时做不同事情的能力
专业术语：
并发：不同的代码块交替执行
并行：不同的代码块同时执行

当第一个线程处于输入输出阶段时，GIL release第一个线程，开始第二个线程，任务交给第二个线程。然后第二个线程用GIL锁住，不让其他线程进行运算。依次循环往复，仅仅是节省了读写的时间（进入读写 --> 切换进程）。

多个线程处理不同的任务效率提升明显（比如：一个线程负责聊天文字读取，一个线程负责文字输出）。
多个线程处理一个相同的任务是没什么提升。如果要提升，需要multiprocessing，多核运不同核之间不会受到GIL的影响

Python的多线程在多核CPU上，只对于IO密集型计算产生正面效果；而当有至少有一个CPU密集型线程存在，那么多线程效率会由于GIL而大幅下降。

避免受到GIL锁的影响
使用multiprocessing，多进程而不是多线程。每个进程有自己的独立的GIL，因此也不会出现进程之间的GIL争抢。
但是它的引入会增加程序实现时线程间数据通讯和同步的困难，multiprocess由于进程之间无法看到对方的数据。

锁

锁住第一个线程，等其执行完，再开始第二个线程。一般当要对多个线程的共同内存处理时，使用锁，完成对共同数据的多步加工。

import threading
def job1():
    global A, lock
    lock.acquire()
    for i in range(10):
        A += 1
        print('job1', A)
    lock.release()
def job2():
    global A, lock
    lock.acquire()
    for i in range(10):
        A += 10
        print('job2', A)
    lock.release()

if __name__ == '__main__':
    lock = threading.Lock()	# 调用Lock
    A = 0   # 全局变量，两个线程的公用内存
    t1 = threading.Thread(target=job1)
    t2 = threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

lock.acquire() 到 lock.release() 之间线程执行的代码时，不会允许其他线程执行。所以之间的代码执行完，才会轮到其他线程执行。

Python多进程/多核 Multiprocessing

利用多核处理器，真正完成同时运行不同的程序。

创建进程

进程的创建与线程完全一样。需要在 if __name__ == '__main__': 中创建。

import multiprocessing as mp
import threading as td

def job(a, b):
    print(a+b)

if __name__ == '__main__':
    t1 = td.Thread(target = job, args = (1, 2))
    p1 = mp.Process(target = job, args = (1, 2))

    t1.start()
    p1.start()
    t1.join()
    p1.join()
1
2
3
4
5
6
7
8
9
10
11
12
13
14

同样需要使用 queue 来储存输出

import multiprocessing as mp

def job(q):
	res = 0
	pass
	q.put(res)
	
q = mp.Queue()
p1 = mp.Process(target = job, args = (q,)) # 如果只有一个参数，要留一个‘ , ’
res1 = q.get()
1
2
3
4
5
6
7
8
9
10

多线程、多进程对比

import multiprocessing as mp
import threading as td
import time

def job(q):
    res = 0
    for i in range(1000000):
        res += i+i**2+i**3
    q.put(res) # queue

def multicore():
    q = mp.Queue()
    p1 = mp.Process(target=job, args=(q,))
    p2 = mp.Process(target=job, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    res1 = q.get()
    res2 = q.get()
    print('multicore:' , res1+res2)

def normal():
    res = 0
    for _ in range(2):
        for i in range(1000000):
            res += i+i**2+i**3
    print('normal:', res)

def multithread():
    q = mp.Queue()
    t1 = td.Thread(target=job, args=(q,))
    t2 = td.Thread(target=job, args=(q,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    res1 = q.get()
    res2 = q.get()
    print('multithread:', res1+res2)

if __name__ == '__main__':
    st = time.time()
    normal()
    st1= time.time()
    print('normal time:', st1 - st)
    print("")
    multithread()
    st2 = time.time()
    print('multithread time:', st2 - st1)
    print("")
    multicore()
    print('multicore time:', time.time()-st2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

进程池 pool

用pool来完成输入，并储存输出。来替代queue

import multiprocessing as mp

def job(x):
    return x*x

def multicore():
    pool = mp.Pool(processes = 3) # processes代表使用几个核，默认使用所有核
    
    # map:一次把很多个值分配个很多个进程来运算
    res = pool.map(job, range(10))	# map（方法，给方法传入的参数）
    print(res)
    # >>> [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
    
    # apply_async: 一次把一个值放入一个核内运算
    res = pool.apply_async(job, (2, ))	# 只能输入一个值
    print(res.get())
    # >>> 4
    # 使用迭代器可以完成多个输入
    multi_res = [pool.apply_async(job, (i, )) for i in range(10)]
    print([res.get() for res in multi_res]
    # >>> [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
    
if __name__ == '__main__':
    multicore()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

共享内存

中间的L2 Cache就是上图四核CPU的共享内存。

import multiprocessing as mp

# 这个value可以被所有进程所共享
value = mp.Value('d', 1)

# 这个一维array可以被所有进程共享
array = mp.Array('i', [1, 2, 3])    # 只能是一维的array
1
2
3
4
5
6
7

通过这些共享内存，计算机的不同核可以相互交流。

进程锁

为了避免进程在共享内存中抢夺，用法同线程的锁。

import multiprocessing as mp

def job(v, num, l):
    l.acquire()
    for _ in range(10):
        time.sleep(0.1)
        v.value += num
        print(v.value)
    l.release()

def multicore():
    l = mp.Lock()
    v = mp.Value('i', 0)
    p1 = mp.Process(target = job, args = (v, 1, l))
    p2 = mp.Process(target = job, args = (v, 3, l))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == '__main__':
    multicore()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

进程之间不互相干扰，p1执行完才执行p2