八、线程和进程

什么是线程（thread）？

线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。

一个线程是一个执行上下文，它是一个CPU需要执行一系列指令的所有信息。
假设你正在读一本书，你现在想休息一下，但是你希望能够回来，从你停止的地方继续阅读。实现这一目标的一种方法是记下页码、行号和字号。所以你读一本书的执行上下文是这三个数字。
如果你有一个室友，而且她使用的是同样的技术，她可以在你不用的时候拿着书，然后从她停止的地方继续阅读。然后你可以把它拿回来，从你所在的地方重新开始。
线程的工作方式相同。一个CPU给你的错觉是它同时在做多个计算。它通过在每个计算上花费一点时间。它可以这样做，因为它对每个计算都有一个执行上下文。就像你可以和你的朋友分享一本书一样，许多任务可以共享一个CPU。
在技术层面上，执行上下文(因此是一个线程)由CPU寄存器的值组成。
最后:线程与进程不同。线程是执行的上下文，而进程是与计算相关的一堆资源。一个进程可以有一个或多个线程。
说明:与进程相关的资源包括内存页(进程中的所有线程都具有相同的内存视图)、文件描述符(例如，打开的套接字)和安全凭据(例如启动进程的用户的ID)。

什么是进程（process）？

程序的执行实例称为进程。
每个进程提供执行程序所需的资源。进程具有虚拟地址空间、可执行代码、对系统对象的打开句柄、安全上下文、惟一进程标识符、环境变量、优先级、最小和最大工作集大小，以及至少一个执行线程。每个进程都是从一个线程开始的，通常称为主线程，但是可以从它的任何线程创建额外的线程。

程序并不能单独运行，只有将程序装载到内存中，系统为它分配资源才能运行，而这种执行的程序就称之为进程。程序和进程的区别就在于：程序是指令的集合，它是进程运行的静态描述文本；进程是程序的一次执行活动，属于动态概念。

在多道编程中，我们允许多个程序同时加载到内存中，在操作系统的调度下，可以实现并发地执行。这是这样的设计，大大提高了CPU的利用率。进程的出现让每个用户感觉到自己独享CPU，因此，进程就是为了在CPU上实现多道编程而提出的。

 线程和进程的区别？

1. 线程共享创建它的进程的地址空间;进程有自己的地址空间。
2. 线程可以直接访问其进程的数据段;进程有自己的父进程数据段的副本。
3. 线程可以直接与进程的其他线程通信;进程必须使用进程间通信来与同胞进程通信。
4. 新线程很容易创建;新进程需要父进程的重复。
5. 线程可以对相同进程的线程进行相当大的控制;进程只能对子进程进行控制。
6. 对主线程的更改(取消、优先级更改等)可能会影响进程的其他线程的行为;对父进程的更改不会影响子进程。

Python GIL(Global Interpreter Lock)全局解释器锁

在CPython中，全局解释器锁(或GIL)是一个互斥锁，可以防止多个本机线程同时执行Python字节码。这一锁定主要是因为CPython的内存管理不是线程安全的。(然而，由于GIL的存在，其他特征已经发展到依赖于它所赋予的保证。)

这里要先明确一点：GIL并不是Python的特性，Python完全可以不依赖于GIL

---

 Python threading（线程）模块

直接调用：

import threading
import time
 
def func(n): #定义每个线程要运行的函数
 
    print("running on number:%s" % n)
 
    time.sleep(3)
 
if __name__ == '__main__':
 
    t1 = threading.Thread(target=func,args=(1,)) #生成一个线程实例
    t2 = threading.Thread(target=func,args=(2,)) #生成另一个线程实例
 
    t1.start() #启动线程
    t2.start() #启动另一个线程
 
    print(t1.getName()) #获取线程名
    print(t2.getName())

继承式调用：

import threading
import time
 
 
class MyThread(threading.Thread):
    def __init__(self,n):
        threading.Thread.__init__(self)
        self.num = n
 
    def run(self):#定义每个线程要运行的函数
 
        print("running on number:%s" %self.n)
 
        time.sleep(2)
 
if __name__ == '__main__':
 
    t1 = MyThread(1)
    t2 = MyThread(2)
    t1.start()
    t2.start()

有些线程执行后台任务，比如发送keepalive数据包，或者执行周期性的垃圾收集，等等。这些只在主程序运行时有用，并且可以在其他非守护进程退出时将它们删除。
如果没有守护线程，您必须跟踪它们，并告诉它们退出，然后程序才能完全退出。通过将它们设置为守护线程，您可以让它们运行并忘记它们，当程序退出时，任何守护线程都会自动被杀死。

def run(n):
    print('[%s] running on
' % n)
    time.sleep(2)
    print('--done--')


def main():
    for i in range(5):
        t = threading.Thread(target=run, args=[i, ])
        t.start()
        t.join(1)
        print('starting thread', t.getName())


m = threading.Thread(target=main, args=[])
# 将main线程设置为Daemon线程,它做为程序主线程的守护线程,当主线程退出时,m线程也会退出,由m启动的其它子线程会同时退出,不管是否执行完任务
m.setDaemon(True) 
m.start()
m.join(timeout=2)
print("---main thread done----")

 

start            线程准备就绪，等待CPU调度
setName          为线程设置名称
getName          获取线程名称
setDaemon        设置为后台线程或前台线程（默认）
                    如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止
                    如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止
join             逐个执行每个线程，执行完毕后继续往下执行
run              线程被cpu调度后自动执行线程对象的run方法

注意：守护程序线程突然停在关闭。它们的资源(如打开文件、数据库事务等)可能无法正常释放。如果您希望您的线程优雅地停止，那么让它们成为非daemonic，并使用一个适当的信号机制，例如事件。

线程锁（Lock、RLock（递归锁））

一个进程下可以启动多个线程，多个线程共享父进程的内存空间，也就意味着每个线程可以访问同一份数据，此时，如果2个线程同时要修改同一份数据，就会得到错误的数据，所以，出现了线程锁 - 同一时刻允许一个线程执行操作。

*注：我用的Python3，不知为什么，结果总是正确的，网上搜了搜，说可能是自动加了锁：以后可能会用Python2版本，也标注一下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import time
import threading
 
def addNum():
    global num #在每个线程中都获取这个全局变量
    print('--get num:',num )
    time.sleep(1)
    num  +=1 #对此公共变量进行-1操作
 
num = 0  #设定一个共享变量
thread_list = []
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
 
for t in thread_list: #等待所有线程执行完毕
    t.join()
 
 
print('final num:', num )

不加锁的听说Python2每次运行结果不都是100.

import time
import threading
 
def addNum():
    global num #在每个线程中都获取这个全局变量
    print('--get num:',num )
    time.sleep(1)
    lock.acquire() #修改数据前加锁
    num  +=1 #对此公共变量进行-1操作
    lock.release() #修改后释放
 
num = 0  #设定一个共享变量
thread_list = []
lock = threading.Lock() #生成全局锁
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
 
for t in thread_list: #等待所有线程执行完毕
    t.join()
 
print('final num:', num )

 

Python已经有一个GIL来保证同一时间只能有一个线程来执行了，为什么这里还需要lock? 注意啦，这里的lock是用户级的lock,跟那个GIL没关系 。

既然用户程序已经自己有锁了，那为什么C python还需要GIL呢？加入GIL主要的原因是为了降低程序的开发的复杂度，比如现在的你写python不需要关心内存回收的问题，因为Python解释器帮你自动定期进行内存回收，你可以理解为python解释器里有一个独立的线程，每过一段时间它起wake up做一次全局轮询看看哪些内存数据是可以被清空的，此时你自己的程序 里的线程和 py解释器自己的线程是并发运行的，假设你的线程删除了一个变量，py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了，结果就有可能新赋值的数据被删除了，为了解决类似的问题，python解释器简单粗暴的加了锁，即当一个线程运行时，其它人都不能动，这样就解决了上述的问题，  这可以说是Python早期版本的遗留问题。

死锁：指多个循环等待他方占有得的资源而无线期僵持下去的局面

死锁
class MyThread(threading.Thread):
    def doA(self):
        lA.acquire()
       # lC.acquire()
        print(self.name,time.ctime())
        time.sleep(2)
        lB.acquire()
       # lC.acquire()
        print(self.name, time.ctime())
        lB.release()
        lA.release()
        #lC.release()
       # lC.release()

    def doB(self):  
        lB.acquire()
        #lC.acquire()
        print(self.name, time.ctime())
        time.sleep(2)
        lA.acquire()
       # lC.acquire()
        print(self.name, time.ctime())
        lA.release()
        lB.release()
        #lC.release()
       # lC.release()

    def run(self):
        self.doA()
        self.doB()

if __name__ == '__main__':
    
    lA = threading.Lock()
    lB= threading.Lock()
   # lC = threading.RLock()
    thread_list = []
    for i in range(5):
        thread_list.append(MyThread())
    for i in thread_list:
        i.start()
    for i in thread_list:
        i.join()

可以用RLock代替Lock

信号量（Semaphore）

互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据

import threading,time
 
def run(n):
    semaphore.acquire()
    time.sleep(1)
    print("run the thread: %s" %n)
    semaphore.release()
 
if __name__ == '__main__':
 
    num= 0
    semaphore  = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
    for i in range(20):
        t = threading.Thread(target=run,args=(i,))
        t.start()

 

Event（事件）

python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。

事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

• clear：将“Flag”设置为False
• set：将“Flag”设置为True

import threading
 
def do(event):
    print('start')
    event.wait()
    print('excute')

event_obj = threading.Event()
for i in range(10):
    t = threading.Thread(target=do,args=(event_obj,))
    t.start()
#event_obj.clear()
inp =input('输入：')
if inp =='true':
    event_obj.set()

通过Event来实现两个或多个线程间的交互：

import time
import random
import threading

def light():
    if not event.is_set():
        event.set() #wait就不阻塞
    count = 0
    while True:
        if count < 10:
            print('33[42;1m--green light on--33[0m')
        elif count < 13:
            print('33[43;1m--yellow light on--33[0m')
        elif count < 20:
            if event.is_set():
                event.clear()
            print('33[41;1m--red light on --33[0m')
        else:
            count = 0
            event.set()
        time.sleep(1)
        count +=1
def car(n):
    while 1:
        time.sleep(random.randrange(10))
        if event.is_set():
            print('car %s is running..' % n)
        else:
            print('car %s is wait')

if __name__ == '__main__':
    event = threading.Event()  #设置事件
    lamp = threading.Thread(target=light) #创建一个灯颜色的线程
    lamp.start()
    for i in range(3):
        t = threading.Thread(target=car,args=(i,))
        t.start()

timer

这个类表示在经过一定时间之后才运行的操作。
计时器的启动方式与线程一样，通过调用它们的start()方法。通过调用thecancel()方法，计时器可以被停止(在它的动作开始之前)。在执行其操作之前，计时器将等待的间隔可能与用户指定的间隔不完全相同。

from threading import Timer
 
 
def hello():
    print("hello, world")
 
t = Timer(1, hello)
t.start()  # after 1 seconds, "hello, world" will be printed

 

条件（Condition）

使得线程等待，只有满足某条件时，才释放n个线程：

import threading
 
def run(n):
    con.acquire()
    con.wait()
    print("run the thread: %s" %n)
    con.release()
 
if __name__ == '__main__':
 
    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()
 
    while True:
        inp = input('>>>')
        if inp == 'q':
            break
        con.acquire()
        con.notify(int(inp))
        con.release()

def condition_func():

    ret = False
    inp = input('>>>')
    if inp == '1':
        ret = True

    return ret


def run(n):
    con.acquire()
    con.wait_for(condition_func) #true就执行下面，不执行
    print("run the thread: %s" %n)
    con.release()

if __name__ == '__main__':

    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()

 queue

当必须在多个线程之间安全地交换信息时，队列在线程编程中特别有用。

class queue.Queue(maxsize=0) #先入先出

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

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

优先队列的构造函数。maxsize是一个整数，它设置可以在队列中放置的条目数的上界限制。一旦达到这个大小，插入将阻塞，直到使用队列项为止。如果maxsize小于或等于0，队列大小是无限的。

最低值的条目首先被检索(最低的值条目是排序后返回的条目)[0]。条目的典型模式是表单中的元组:(priority_number, data)。

exception queue.Empty

当非阻塞get(或get_nowait())被调用在一个空的队列对象上时，异常会发生。

exception queue.Full

非阻塞put()(或put_nowait())被调用在一个已满的队列对象时引发异常。

Queue.qsize()
Queue.empty()    #return True if empty
Queue.full()         # return True if full
Queue.put(item, block=True, timeout=None)

将项目放入队列中。如果可选的args block为真，超时为None(默认)，则在必要时阻塞，直到空闲插槽可用为止。如果超时是一个正数，它会在大多数超时秒内阻塞，如果在那个时间内没有空闲插槽，就会引发完全的异常。否则(block为false)，如果立即可用空闲插槽，则在队列上放置一个项目，否则将引发完全异常(在这种情况下，超时将被忽略)

Queue.put_nowait(item)
Equivalent to put(item, False).

Queue.get(block=True, timeout=None)

从队列中移除并返回一个项。如果可选的args block为真，timeout为None(默认)，则在必要时阻塞，直到项目可用为止。如果超时是一个正数，它会在大多数超时秒内阻塞，如果在那个时间内没有可用的项，则会引发空的异常。否则(块为false)，如果立即可用，则返回一个项，否则将引发空异常(在这种情况下，超时将被忽略)。

Queue.task_done()

指示以前的队列任务已完成。用于队列消费线程。对于用于获取任务的每个get()，对task_done()的后续调用告诉队列，任务的处理是完整的。

如果一个join()当前正在阻塞，那么当所有的项目都被处理后，它将恢复(这意味着对已放入队列中的每一个项目都将收到一个task_done()调用)。

如果调用的次数多于队列中的项，则会产生一个ValueError错误。

Queue.join() block直到队列被消费完毕

常用的：

1. 初始化： class Queue.Queue(maxsize) FIFO 先进先出

2. 包中的常用方法:

   • Queue.qsize() 返回队列的大小

   • Queue.empty() 如果队列为空，返回True,反之False

   • Queue.full() 如果队列满了，返回True,反之False

   • Queue.full 与 maxsize 大小对应

   • Queue.get([block[, timeout]])获取队列，timeout等待时间

3. 创建一个“队列”对象

   • import Queue
   • myqueue = Queue.Queue(maxsize = 10)

4. 将一个值放入队列中

   • myqueue.put(10)

5. 将一个值从队列中取出

   • myqueue.get() 

 什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

import time,random
import queue,threading
q = queue.Queue()
def Producer(name):
  count = 0
  while count <20:
    time.sleep(random.randrange(3))
    q.put(count)
    print('Producer %s has produced %s baozi..' %(name, count))
    count +=1
def Consumer(name):
  count = 0
  while count <20:
    time.sleep(random.randrange(4))
    if not q.empty():
        data = q.get()
        print(data)
        print('33[32;1mConsumer %s has eat %s baozi...33[0m' %(name, data))
    else:
        print("-----no baozi anymore----")
    count +=1
p1 = threading.Thread(target=Producer, args=('A',))
c1 = threading.Thread(target=Consumer, args=('B',))
p1.start()
c1.start()

---

python multiprocessing

multiprocessing是一个包，它支持使用与线程模块类似的API来生成进程。multiprocessing包提供本地和远程并发性，通过使用子进程代替线程，有效地绕过全局解释器锁。由于这个原因，multiprocessing模块允许程序员在给定的机器上充分利用多个处理器。它在Unix和Windows上运行。

from multiprocessing import Process
import time
def f(name):
    time.sleep(2)
    print('hello', name)
 
if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

注意：由于进程之间的数据需要各自持有一份，所以创建进程需要的非常大的开销.

进程的通信：

不同进程间内存是不共享的，要想实现两个进程间的数据交换，可以用以下方法：

Queues

使用方法跟threading里的queue差不多

from multiprocessing import Process, Queue
 
def func(q):
    q.put([1,2,3])
 
if __name__ == '__main__':
    q = Queue()
    p = Process(target=func, args=(q,))
    p.start()
    print(q.get())    # prints "[1,2,3]"
    p.join()

Pipes

函数pipe()的作用是:返回由管道连接的一对连接对象，默认情况下是双向的。例如:

from multiprocessing import Process, Pipe
 
def f(conn):
    conn.send([42, None, 'hello'])
    conn.close()
 
if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # prints "[42, None, 'hello']"
    p.join()

由pipe()返回的两个连接对象表示管道的两端。每个连接对象都有send()和recv()方法。注意，如果两个进程(或线程)试图同时读取或写入管道的同一端口，那么管道中的数据可能会被损坏。当然，在同时使用不同端口的过程中也不会有腐败的风险。

Managers

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by Manager() will support:

types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array. For example,

from multiprocessing import Process, Manager
 
def f(d, l):
    d[1] = '1'
    d['2'] = 2
    d[0.25] = None
    l.append(1)
    print(l)
 
if __name__ == '__main__':
    with Manager() as manager:
        d = manager.dict()
 
        l = manager.list(range(5))
        p_list = []
        for i in range(10):
            p = Process(target=f, args=(d, l))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()
 
        print(d)
        print(l)

进程池

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。

进程池中有两个方法：

• apply
• apply_async

from  multiprocessing import Process,Pool
import time
 
def Foo(i):
    time.sleep(2)
    return i+100
 
def Bar(arg):
    print('-->exec done:',arg)
 
pool = Pool(5)
 
for i in range(10):
    pool.apply_async(func=Foo, args=(i,),callback=Bar)
    #pool.apply(func=Foo, args=(i,))
 
print('end')
pool.close()
pool.join()#进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。