多线程与多核运算（Threading&MultiProcess)_多核和多线程

1.多线程

什么是多线程

线程是操作系统能够进行运算调度的最小单位；它被包含在进程之中，是进程中的实际运作单位。

多线程，是指从软件或者硬件上实现多个线程并发执行的技术。具有多线程能力的计算机因有硬件支持而能够在同一时间执行多于一个线程，进而提升整体处理性能。
1
2
3

简单来说：线程是程序中一个单一的顺序控制流程；而多线程就是在单个程序中同时运行多个线程来完成不同的工作。

多线程是为了同步完成多项任务，不是为了提高运行效率，而是为了提高资源使用效率来提高系统的效率。多线程是在同一时间需要完成多项任务的时候实现的。

多线程的优缺点

优点：

1）多线程技术可以加快程序的运行速度，使程序的响应速度更快，因为用户界面可以在进行其它工作的同时一直处于活动状态

2）可以把占据长时间的程序中的任务放到后台去处理，同时执行其他操作，提高效率

3）当前没有进行处理的任务时可以将处理器时间让给其它任务

4）可以让同一个程序的不同部分并发执行，释放一些珍贵的资源如内存占用等等

5）可以随时停止任务

6）可以分别设置各个任务的优先级以优化性能

缺点：

1）因为多线程需要开辟内存，而且线程切换需要时间因此会很消耗系统内存。

2）线程的终止会对程序产生影响

3）由于多个线程之间存在共享数据，因此容易出现线程死锁的情况

4）对线程进行管理要求额外的 CPU开销。线程的使用会给系统带来上下文切换的额外负担

线程的添加

import threading
def thread_job1():
    print('This is a  new thread! %s\n'%threading.current_thread())

def main():
    #添加一个线程
    add_thread1 = threading.Thread(target=thread_job1)

    #执行此线程
    add_thread1.start()
    #当前线程
    print(threading.enumerate())

    #当前执行线程
    print(threading.current_thread())

    #当前线程数
    print(threading.active_count())
if __name__ == '__main__':
    main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

join功能

join的原理就是依次检验线程池中的线程是否结束，没有结束就阻塞直到线程结束，如果结束则跳转执行下一个线程的join函数。

import threading
import time

def thread_job():
    print('T1 is strat\n')
    for i in range(10):
        time.sleep(0.1)
    print('T1 finish\n')


def thread2_job():
    print('T2 is strat\n')
    print('T2 finish\n')


def main():
    #添加一个线程
    add_thread = threading.Thread(target=thread_job, name='T1')
    add_thread2 = threading.Thread(target=thread2_job,name='T2')
    #执行此线程
    add_thread.start()
    add_thread2.start()

    add_thread2.join()
    add_thread.join()
    print('all done\n')

if __name__ == '__main__':
    main()
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

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)

def multithreading():
    q = Queue()
    threads = []
    data =[[1,2,3],[4,5,6],[7,8,9],[6,6,6]]
    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()
    results =[]
    for _ in range(4):
        results.append(q.get())
    print(results)
if __name__ == '__main__':
    multithreading()
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

GIL

import threading
from queue import Queue
import copy
import time

def job(l, q):
    res = sum(l)
    q.put(res)

def multithreading(l):
    q = Queue()
    threads = []
    for i in range(4):
        t = threading.Thread(target=job, args=(copy.copy(l), q), name='T%i' % i)
        t.start()
        threads.append(t)
    [t.join() for t in threads]
    total = 0
    for _ in range(4):
        total += q.get()
    print(total)

def normal(l):
    total = sum(l)
    print(total)

if __name__ == '__main__':
    l = list(range(1000000))
    s_t = time.time()
    normal(l*4)
    print('normal: ',time.time()-s_t)
    s_t = time.time()
    multithreading(l)
    print('multithreading: ', time.time()-s_t)
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

线程安全与Lock锁

什么是线程安全？
当多个线程访问某个方法时，不管你通过怎样的调用方式、或者说这些线程如何交替地执行，我们在主程序中不需要去做任何的同步，这个类的结果行为都是我们设想的正确行为，那么我们就可以说这个类是线程安全的。
既然是线程安全问题，那么毫无疑问，所有的隐患都是在多个线程访问的情况下产生的，也就是我们要确保在多条线程访问的时候，我们的程序还能按照我们预期的行为去执行。
Lock锁

1. 凡是存在共享资源争抢的地方都可以使用锁，从而保证只有一个使用者可以完全使用这个资源一旦线程获得锁，其他试图获取锁的线程将被阻塞
2. acquire(blocking=True,timeout=-1): 默认阻塞，阻塞可以设置超时时间，非阻塞时，timeout禁止设置，成功获取锁，返回True，否则返回False
3. releas() : 释放锁，可以从任何线程调用释放，已上锁的锁，会被重置为unlocked未上锁的锁上调用，抛出RuntimeError异常

未使用Lock锁：案例

import threading
def job1():
    global Q
    for i in range(10):
        Q+=1
        print('job1 ',Q)
def job2():
    global Q
    for i in range(10):
        Q += 10
        print('job2 ', Q)
if __name__ == '__main__':
    Q = 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

使用Lock锁后：

import threading

def job1():
    global Q,lock
    lock.acquire()
    for i in range(10):
        Q+=1
        print('job1 ',Q)
    lock.release()
def job2():
    global Q,lock
    lock.acquire()
    for i in range(10):
        Q += 10
        print('job2 ', Q)
    lock.release()
if __name__ == '__main__':
    Q = 0
    lock = threading.Lock()
    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
26
27
28
29

2.多进程

什么是多进程

进程是程序在计算机上的一次执行活动。当你运行一个程序，你就启动了一个进程。凡是用于完成操作系统的各种功能的进程就是系统进程，而所有由你启动的进程都是用户进程。多进程就是指计算机同时执行多个进程，一般是同时运行多个软件。

创建多进程

import multiprocessing as mp
import threading as td
def job(a,d):
    print("aaaa")

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

    p1.start()

    p1.join()

1
2
3
4
5
6
7
8
9
10
11
12

queue

不同进程间内存是不共享的，要想实现两个进程间的数据交换，可以使用Queue，类似与Threading中的Queue

import multiprocessing as mp

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

if __name__ == '__main__':
    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(res1+res2)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

进程池Pool

为什么要有进程池?
在程序实际处理问题过程中，忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？首先，创建进程需要消耗时间，销毁进程也需要消耗时间。第二即便开启了成千上万的进程，操作系统也不能让他们同时执行，这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。那么我们要怎么做呢？

在这里，定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果。

import multiprocessing as mp

def job(x):
    return x*x

def multicore():

    #自动分配
    pool = mp.Pool(processes=3)
    #pool = mp.Pool(processes=3)
    #processes表示固定核数
    res = pool.map(job,range(10))
    print(res)

    #单进程处理
    res = pool.apply_async(job,(2,))
    print(res.get())

    #单进程迭代
    multi_res = [pool.apply_async(job,(i,)) for i in range(10)]
    print([res.get() for res in multi_res])




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
25
26
27

Share Memory与Lock锁

import  multiprocessing as mp
import time
#进程数据交流
#共享内存数据 d 表示数据类型



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

def multiProcesssing_job():
    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__':
    multiProcesssing_job()
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

3.多线程与多进程的选择

多线程与多进程，选择谁？
下面是本人从知乎上转载的一个答案，非常通俗地回答了这个问题。

1. 单进程单线程：一个人在一个桌子上吃菜。
2. 单进程多线程：多个人在同一个桌子上一起吃菜。
3. 多进程单线程：多个人每个人在自己的桌子上吃菜。

开桌子的意思是指创建进程。开销这里主要指的是时间开销。
多线程的问题是多个人同时吃一道菜的时候容易发生争抢，例如两个人同时夹一个菜，一个人刚伸出筷子，结果伸到的时候已经被夹走菜了。。。此时就必须等一个人夹一口之后，在还给另外一个人夹菜，也就是说资源共享就会发生冲突争抢。

1. 对于 Windows 系统来说，【开桌子】的开销很大，因此 Windows 鼓励大家在一个桌子上吃菜。因此 Windows 多线程学习重点是要大量面对资源争抢与同步方面的问题。

2. 对于 Linux 系统来说，【开桌子】的开销很小，因此 Linux 鼓励大家尽量每个人都开自己的桌子吃菜。这带来新的问题是：坐在两张不同的桌子上，说话不方便。因此，Linux 下的学习重点大家要学习进程间通讯的方法。

多线程与多进程的对比

import multiprocessing as mp
import threading as td
import time
def normal():
    res = 0
    for _ in range(2):
        for i in range(100000):
            res += i + i ** 2 + i ** 3
    print('normal ',res)

def threading_job():
    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('threading ',res1 + res2)
def multiProcesssing_job():
    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('multiProcessing ',res1 + res2)
def job(q):
    res = 0
    for i in  range(100000):
        res += i+i**2+i**3
    q.put(res)



if __name__ == '__main__':
   st = time.time()
   normal()
   st1 =time.time()
   print('normal ',st1-st)

   threading_job()
   st2 = time.time()
   print('threading_job ', st2 - st1)

   multiProcesssing_job()
   st3 = time.time()
   print('multiProcesssing_job ', st3 - 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
54
55
56
57
58
59
60