并发编程只多线程

线程理论

一什么是线程

在传统操作系统中，每个进程有一个地址空间，而且默认就有一个控制线程

所以，进程只是用来把资源集中到一起（进程只是一个资源单位，或者说资源集合），而线程才是cpu上的执行单位。

多线程（即多个控制线程）的概念是，在一个进程中存在多个线程，多个线程共享该进程的地址空间，相当于一个车间内有多条流水线，都共用一个车间的资源

二线程与进程的区别

同一个进程内的多个线程共享该进程内的地址资源
创建线程的开销要远小于创建进程的开销（创建一个进程，就是创建一个车间，涉及到申请空间，而且在该空间内建至少一条流水线，但创建线程，就只是在一个车间内造一条流水线，无需申请空间，所以创建开销小）

开启线程的两种方式

一 threading模块介绍

multiprocess模块的完全模仿了threading模块的接口，二者在使用层面，有很大的相似性，因而不再详细介绍。

方式一

 1 from threading import Thread
 2 import time
 3 
 4 def sayhi(name):
 5     time.sleep(2)
 6     print('%s say hello' %name)
 7 
 8 if __name__ == '__main__':
 9     t=Thread(target=sayhi,args=('egon',))
10     t.start()
11     print('主线程')

View Code

方式二

 1 #方式二
 2 from threading import Thread
 3 import time
 4 
 5 class Sayhi(Thread):
 6     def __init__(self,name):
 7         super().__init__()
 8         self.name=name
 9     def run(self):
10         time.sleep(2)
11         print('%s say hello' % self.name)
12 
13 if __name__ == '__main__':
14     t = Sayhi('egon')
15     t.start()
16     print('主线程')

View Code

运行代码，开启一个主进程，主进程中的主线程通过t=Thread(target=sayhi,args=('egon',))，t.start()开启了另一个线程

多线程与多进程

同一进程内的线程都是平级的，不存在子线程之说，只有进程才存在主进程和子进程之说

开启进程，操作系统要申请内存空间，让好拷贝父进程地址空间到子进程，开销远大于线程。

from threading import Thread

def work():
    print('hello')

if __name__ == '__main__':
    t=Thread(target=work)
    t.start()
    print('主线程/主进程')

hello
主线程/主进程

执行结果如下，几乎是t.start ()的同时就将线程开启了，然后先打印出了hello，证明线程的创建开销极小。

同一进程内的线程pid一样，不同进程pid不同

线程：

from threading import Thread
import os

def work():
    print('hello',os.getpid())

if __name__ == '__main__':
    t1=Thread(target=work)
    t2=Thread(target=work)
    t1.start()
    t2.start()
    print('主线程/主进程pid',os.getpid())

hello 7939
hello 7939
主线程/主进程 7939

　　进程：

from multiprocessing import Process
import os

def work():
    print('hello',os.getpid())

if __name__ == '__main__':
    p1=Process(target=work)
    p2=Process(target=work)
    p1.start()
    p2.start()
    print('主线程/主进程',os.getpid())

主线程/主进程 7951
hello 7952
hello 7953

多线程的执行本质上来说还是一个个线程按顺序来执行，但基于多道技术，后面发起的线程不需要等待前面的线程完全结束后才开始执行，当前面的线程遇到io阻塞或占用cpu时间过长时系统会自动切到其他的线程去（多道技术）；

而之前我们所写的程序就是按顺序来执行的，执行了一个函数（以上例子的子线程也是调用函数）后，必须等待这个函数的结束return后才能执行下一个函数（无论这个函数中间有没有io阻塞或占用cpu时间过长），而多道技术不需要等到一个函数结束才能执行其他函数，依据多道技术系统会自动io阻塞和cpu占用时间来切换，可以类似于我们之前在写函数时在函数内部加一个条件，当遇到这个条件时去调用其他函数，这样实现和一个函数切换到另一个函数，但没有实现再切换回来，多道技术不仅可以让一个函数（线程）切换到另一个函数（线程），还可以再切回到原来的函数继续执行；

综上所述，当我们能实现将函数切换到另一个函数再切回来就可以模拟（多道技术）并发操作，在后面将IO模型，我们将讲述如何模拟多道技术中遇到io阻塞时切换的情况（没有模拟占用cpu时间过长的情况）。

Thread对象的其他属性或方法

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

threading模块提供的一些方法：
  # threading.currentThread(): 返回当前的线程变量。
  # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果

 1 from threading import Thread
 2 import threading
 3 from multiprocessing import Process
 4 import os
 5 
 6 def work():
 7     import time
 8     time.sleep(3)
 9     print(threading.current_thread().getName())
10 
11 
12 if __name__ == '__main__':
13     #在主进程下开启线程
14     t=Thread(target=work)
15     t.start()
16 
17     print(threading.current_thread().getName())
18     print(threading.current_thread()) #主线程
19     print(threading.enumerate()) #连同主线程在内有两个运行的线程
20     print(threading.active_count())
21     print('主线程/主进程')

View Code

MainThread
<_MainThread(MainThread, started 15728)>
[<_MainThread(MainThread, started 15728)>, <Thread(Thread-1, started 16556)>]
2
主线程/主进程
Thread-1

守护线程

守护线程会等待主线程运行完毕后被销毁

注意：运行完毕并非终止运行，运行完毕如下

　　1、对主进程来说，运行完毕指的是主进程代码运行完毕（主进程内的所有线程都运行完毕），和子进程无关，终止运行则需要等待子进程也运行完毕，资源回收后才会终止运行；

　　2、对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕（主线程运行完毕也是主进程运行完毕），但主进程结束需要等到所有子进程的结束，资源回收后才结束（主进程结束也伴随着子进程结束）

 1 from threading import Thread
 2 import time
 3 def sayhi(name):
 4     time.sleep(2)
 5     print('%s say hello' %name)
 6 
 7 if __name__ == '__main__':
 8     t=Thread(target=sayhi,args=('egon',))
 9     t.setDaemon(True) #必须在t.start()之前设置
10     t.start()
11 
12     print('主线程')
13     print(t.is_alive())

View Code

主线程
True

GIL全局解释锁

 1 from threading import Thread
 2 import time
 3 
 4 n = 100
 5 
 6 def task():
 7     global n
 8     temp = n
 9     time.sleep(0.1)
10     n = temp - 1
11 
12 if __name__ == '__main__':
13     t_l = []
14     for i in range(100):
15         t = Thread(target=task)
16         t_l.append(t)
17         t.start()
18     for t in t_l:
19         t.join()
20     print('主',n)

View Code

主 99

　　结果n并不是0，而是99，因为在n减1之前，99个线程都开始了，拿到的n的值都是100,因此当所有线程计算结束时n仍然是99。

加锁：

 1 from threading import Thread,Lock
 2 import os,time
 3 def work():
 4     global n
 5     lock.acquire()
 6     temp=n
 7     time.sleep(0.1)
 8     n=temp-1
 9     lock.release()
10 if __name__ == '__main__':
11     lock=Lock()
12     n=100
13     l=[]
14     for i in range(100):
15         p=Thread(target=work)
16         l.append(p)
17         p.start()
18     for p in l:
19         p.join()
20 
21     print(n)

View Code

死锁与递归锁

一死锁现象

所谓死锁：是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁。

 1 class MyThread(Thread):
 2     def run(self):
 3         self.func1()
 4         self.func2()
 5     def func1(self):
 6         mutexA.acquire()
 7         print('33[41m%s 拿到A锁33[0m' %self.name)
 8 
 9         mutexB.acquire()
10         print('33[42m%s 拿到B锁33[0m' %self.name)
11         mutexB.release()
12 
13         mutexA.release()
14 
15     def func2(self):
16         mutexB.acquire()
17         print('33[43m%s 拿到B锁33[0m' %self.name)
18         time.sleep(2)
19 
20         mutexA.acquire()
21         print('33[44m%s 拿到A锁33[0m' %self.name)
22         mutexA.release()
23 
24         mutexB.release()
25 
26 if __name__ == '__main__':
27     for i in range(10):
28         t=MyThread()
29         t.start()

View Code

Thread-1 拿到A锁
Thread-1 拿到B锁
Thread-1 拿到B锁
Thread-2 拿到A锁 #出现死锁，整个程序阻塞住

二递归锁

递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁。

二者的区别是：递归锁可以连续acquire多次，而互斥锁只能acquire一次。

 1 from threading import Thread,RLock
 2 import time
 3 
 4 mutexA=mutexB=RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止
 5 
 6 class MyThread(Thread):
 7     def run(self):
 8         self.func1()
 9         self.func2()
10     def func1(self):
11         mutexA.acquire()
12         print('33[41m%s 拿到A锁33[0m' %self.name)
13 
14         mutexB.acquire()
15         print('33[42m%s 拿到B锁33[0m' %self.name)
16         mutexB.release()
17 
18         mutexA.release()
19 
20     def func2(self):
21         mutexB.acquire()
22         print('33[43m%s 拿到B锁33[0m' %self.name)
23         time.sleep(2)
24 
25         mutexA.acquire()
26         print('33[44m%s 拿到A锁33[0m' %self.name)
27         mutexA.release()
28 
29         mutexB.release()
30 
31 if __name__ == '__main__':
32     for i in range(10):
33         t=MyThread()
34         t.start()

View Code

信号量，event，定时器

一信号量

也是锁，但多个用户可以同时拥有这把锁，即可以允许多个用户同时共享同一资源，允许的用户数由参数指定；

 1 rom threading import Thread,Semaphore
 2 import threading
 3 import time
 4 
 5 def func():
 6     sm.acquire()
 7     print('%s get sm' %threading.current_thread().getName())
 8     time.sleep(3)
 9     sm.release()
10 
11 if __name__ == '__main__':
12     sm=Semaphore(5)
13     for i in range(23):
14         t=Thread(target=func)
15         t.start()

View Code

Semaphore管理一个内置的计数器，该计数器由参数指定。
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

二 Event

在程序中使用Threading库中的Event对象可以让其他线程（多个线程）通过某一线程（一个线程）的状态来确定自己下一步的操作。

在初始情况下,Event对象中的信号标志被设置为假。
有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。
一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。
如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行。

from threading import Event

event.isSet()：返回event的状态值；

event.wait()：如果 event.isSet()==False将阻塞线程，可设置超时时间，当阻塞超过此时间时线程会继续往下走

event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；

event.clear()：恢复event的状态值为False。

 1 from threading import Thread,Event
 2 import threading
 3 import time
 4 
 5 event = Event()
 6 
 7 def student(name):
 8     print('学生%s 正在听课'%name)
 9     event.wait()
10     event.wait(1) #不需要等到event.set()发送Ture信号，在等待1s后线程就会继续往下运行
11     print('学生%s 课间活动'%name)
12 
13 def teacher(name):
14     print('老师%s 正在授课'%name)
15     time.sleep(5)
16     event.set()
17 
18 if __name__=='__main__':
19     stu1 = Thread(target=student,args=('hh',))
20     stu2 = Thread(target=student,args=('oo',))
21     stu3 = Thread(target=student,args=('ww',))
22     t1 = Thread(target=teacher,args=('qq',))
23 
24     stu1.start()
25     stu2.start()
26     stu3.start()
27     t1.start()

View Code

线程queue

queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.

有三种不同的用法

class queue.Queue(maxsize=0) #队列：先进先出

 1 import queue
 2 
 3 q=queue.Queue()
 4 q.put('first')
 5 q.put('second')
 6 q.put('third')
 7 
 8 print(q.get())
 9 print(q.get())
10 print(q.get())

View Code

'''
结果(后进先出):
third
second
first
'''

class queue.LifoQueue(maxsize=0) #堆栈：last in fisrt out

 1 import queue
 2 
 3 q=queue.LifoQueue()
 4 q.put('first')
 5 q.put('second')
 6 q.put('third')
 7 
 8 print(q.get())
 9 print(q.get())
10 print(q.get())

View Code

结果(后进先出):
third
second
first

　　class queue.PriorityQueue(maxsize=0) #优先级队列：存储数据时可设置优先级的队列

 1 import queue
 2 
 3 q=queue.PriorityQueue()
 4 #put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
 5 q.put((20,'a'))
 6 q.put((10,'b'))
 7 q.put((30,'c'))
 8 
 9 print(q.get())
10 print(q.get())
11 print(q.get())

View Code

结果(数字越小优先级越高,优先级高的优先出队):
(10, 'b')
(20, 'a')
(30, 'c')

进程和进程池

一进程池与线程池

在刚开始学多进程或多线程时，我们迫不及待地基于多进程或多线程实现并发的套接字通信，然而这种实现方式的致命缺陷是：服务的开启的进程数或线程数都会随着并发的客户端数目地增多而增多，这会对服务端主机带来巨大的压力，甚至于不堪重负而瘫痪，于是我们必须对服务端开启的进程数或线程数加以控制，让机器在一个自己可以承受的范围内运行，这就是进程池或线程池的用途，例如进程池，就是用来存放进程的池子，本质还是基于多进程，只不过是对开启进程的数目加上了限制。

介绍

官网：https://docs.python.org/dev/library/concurrent.futures.html

concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.

基本方法

1、submit(fn, *args, **kwargs)
异步提交任务

2、map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

3、shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

4、result(timeout=None)
取得结果

5、add_done_callback(fn)
回调函数

二进程池

 1 from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
 2 
 3 import os,time,random
 4 def task(n):
 5     print('name：%s pid：%sis runing' %(n,os.getpid()))
 6     time.sleep(random.randint(1,3))
 7     return n
 8 
 9 if __name__ == '__main__':
10 
11     pool=ProcessPoolExecutor(max_workers=3)
12 
13     futures=[]
14     for i in range(11):
15         future=pool.submit(task,'hh%s'%i)
16         futures.append(future)
17 
18     pool.shutdown(wait=True)
19     '''
20     相当于进程池的pool.close()+pool.join()操作,必须先关闭进程池，因为如果未关闭，则可能在等待进程结束的过程中有新的进程开启
21     wait=True，等待池内所有任务执行完毕回收完资源后才继续
22     wait=False，立即返回，并不会等待池内的任务执行完毕
23     但不管wait参数为何值，整个程序都会等到所有任务执行完毕
24     submit和map必须在shutdown之前
25     '''
26     for future in futures:
27         print(future.result()) 
28     print('主')

View Code

name：hh0 pid：9176is runing
name：hh1 pid：10872is runing
name：hh2 pid：5732is runing
name：hh3 pid：9176is runing
name：hh4 pid：10872is runing
name：hh5 pid：5732is runing
name：hh6 pid：10872is runing
name：hh7 pid：9176is runing
name：hh8 pid：5732is runing
name：hh9 pid：9176is runing
name：hh10 pid：10872is runing
hh0
hh1
hh2
hh3
hh4
hh5
hh6
hh7
hh8
hh9
hh10
主

三线程池

用法

把ProcessPoolExecutor换成ThreadPoolExecutor，其余用法全部相同


    同步调用：提交任务后 就在原地等待任务执行完毕，拿到结果，再执行下一行代码，导致任务是串行执行

 1 from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
 2 import os,time,random
 3 
 4 def get(name):
 5     print('%s is getting'%name)
 6     time.sleep(random.randint(3,5))
 7     res = random.randint(7,13)*'#'
 8     return {'name':name,'res':res}
 9 
10 def weigh(shit):
11 
12     name = shit['name']
13     size = len(shit['res'])
14     print('%s 得到了（%s）kg'%(name,size))
15 
16 if __name__ == '__main__':
17     pool = ThreadPoolExecutor(13)
18     shit1 = pool.submit(get,'小明').result() #result() 必须等待线程结束后才能执行下一行代码,因此代码是串行执行
19     weigh(shit1)
20     shit2 = pool.submit(get,'小红').result()
21     weigh(shit2)
22     shit3 = pool.submit(get,'小白').result()
23     weigh(shit3)

View Code

小明 is getting
小明 得到了（12）kg
小红 is getting
小红 得到了（12）kg
小白 is getting
小白 得到了（11）kg

　　异步调用：提交完任务后 ，不需要等待任务执行完毕。
    用到回调函数

 1 from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
 2 import os,time,random
 3 
 4 def get(name):
 5     print('%s is getting'%name)
 6     time.sleep(random.randint(3,5))
 7     res = random.randint(7,13)*'#'
 8     return {'name':name,'res':res}
 9 
10 def weigh(shit):
11     shit = shit.result()
12     name = shit['name']
13     size = len(shit['res'])
14     print('%s 得到了（%s）kg'%(name,size))
15 
16 if __name__ == '__main__':
17     pool = ThreadPoolExecutor(13)
18     shit1 = pool.submit(get,'小明').add_done_callback(weigh)
19     #add_done_callback(weigh)为线程绑定一个函数，该函数在线程任务执行完毕后自动触发，并接收线程任务当作参数
20     shit2 = pool.submit(get,'小红').add_done_callback(weigh)
21     shit3 = pool.submit(get,'小白').add_done_callback(weigh)

View Code

小明 is getting
小红 is getting
小白 is getting
小明 得到了（9）kg
小白 得到了（11）kg
小红 得到了（9）kg