Python之路-Python中的线程与进程

一.发展背景

　　任务调度

　　大部分操作系统(如Windows、Linux)的任务调度是采用时间片轮转的抢占式调度方式，也就是说一个任务执行一小段时间后强制暂停去执行下一个任务，每个任务轮流执行。任务执行的一小段时间叫做时间片，任务正在执行时的状态叫运行状态，任务执行一段时间后强制暂停去执行下一个任务，被暂停的任务就处于就绪状态等待下一个属于它的时间片的到来。这样每个任务都能得到执行，由于CPU的执行效率非常高，时间片非常短，在各个任务之间快速地切换，给人的感觉就是多个任务在“同时进行”，这也就是我们所说的并发(一个时间间隔内,多个任务同时进行).

　　进程

　　计算机的核心是CPU，承担所有计算任务；操作系统是计算机管理者，负责任务的调度、资源的分配和管理，统领整个计算机硬件；应用程序侧是具有某种功能的程序，程序是运行于操作系统之上的。

　　进程是一个具有一定独立功能的程序在一个数据集上的一次动态执行的过程，是操作系统进行资源分配和调度的一个独立单位，是应用程序运行的载体。

　　进程是一种抽象的概念，从来没有统一的标准定义。进程一般由程序、数据集合和进程控制块三部分组成。程序用于描述进程要完成的功能，是控制进程执行的指令集；数据集合是程序在执行时所需要的数据和工作区；程序控制块(Program Control Block，简称PCB)，包含进程的描述信息和控制信息，是进程存在的唯一标志。

　　进程具有的特征：

　　动态性：进程是程序的一次执行过程，是临时的，有生命期的，是动态产生，动态消亡的；

　　并发性：任何进程都可以同其他进程一起并发执行；

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

　　结构性：进程由程序、数据和进程控制块三部分组成。

　　线程

　　早期的操作系统中并没有线程的概念，进程是能拥有资源和独立运行的最小单位，也是程序执行的最小单位。任务调度采用的是时间片轮转的抢占式调度方式，而进程是任务调度的最小单位，每个进程有各自独立的一块内存，使得各个进程之间内存地址相互隔离。 后来，随着计算机的发展，对CPU的要求越来越高，进程之间的切换开销较大，已经无法满足越来越复杂的程序的要求了。于是就发明了线程，线程是程序执行中一个单一的顺序控制流程，是程序执行流的最小单元，是处理器调度和分派的基本单位.一个进程可以有一个或多个线程，各个线程之间共享程序的内存空间(也就是所在进程的内存空间)。一个标准的线程由线程ID、当前指令指针(PC)、寄存器和堆栈组成。而进程由内存空间(代码、数据、进程空间、打开的文件)和一个或多个线程组成。

二.进程与线程的区别

　　1.线程是程序执行的最小单位，而进程是操作系统分配资源的最小单位；

　　2.一个进程由一个或多个线程组成，线程是一个进程中代码的不同执行路线；

　　3.进程之间相互独立，但同一进程下的各个线程之间共享程序的内存空间(包括代码段、数据集、堆等)及一些进程级的资源(如打开文件和信号)，某进程内的线程在其它进程不可见；

　　4.调度和切换：线程上下文切换比进程上下文切换要快得多。

总之，线程和进程都是一种抽象的概念，线程是一种比进程更小的抽象，线程和进程都可用于实现并发。

 

　　单线程

　　在单线程中，当处理器需要处理多个任务时，必须对这些任务安排执行顺序，并按照这个顺序来执行任务。假如我们创建了两个任务：听音乐（music)和看电影（movie)。在单线程中，我们只能按先后顺序来执行这两个任务。

 

import time
import threading


#单线程
'''
def music(name,loop):
    for i in range(loop):
        print('listen music %s %s'%(name,time.ctime()))
        time.sleep(1)

def movie(name,loop):
    for i in range(loop):
        print('listen movie %s %s'%(name,time.ctime()))
        time.sleep(1)

if __name__== '__main__':
    music('爱的故事上集',3)
    movie('肖生克的救赎',4)
    print('end time %s' %time.ctime())
'''

 

三.python多线程模块threading

　　Python中的多线程

　　Python中存在多线程，但是很遗憾，python仅仅只支持一个线程的运行（原因如下）

　　thread模块

　　threading模块

　　推荐使用threading模块

　　主要原因是， thread不支持守护线程。当主线程退出时，所有的子线程不管它们是否还在工作，都会被强行退出。有时我们并不希望发生这种行为，这时就引入了守护线程的概念。threading模块支持守护线程

　　Python多线程基本步骤

　　threading使用简介-1

　　threading.Thread(self, group=None, target=None, name=None, args=(), kwargs={}) 　　

　　Thread 是threading模块中最重要的类之一，可以使用它来创建线程。有两种方式来创建线程：一种是通过继承Thread类，重写它的run方法；另一种是创建一个threading.Thread对象，在它的初始化函数（__init__）中将可调用对象作为参数传入。

　　参数group是预留的，用于将来扩展；

　　参数target是一个可调用对象（也称为活动[activity]），在线程启动后执行；

　　参数name是线程的名字。默认值为“Thread-N“，N是一个数字。

　　参数args和kwargs分别表示调用target时的参数列表和关键字参数。

 　　threading使用简介-2

　　Thread类还定义了以下常用方法与属性：

　　Thread.getName()/ Thread.name 用于获取线程的名称,有返回值,需要print打印出来

　　Thread.setName() 用于设置线程的名称,没有返回值,print打印出来事None

　　Thread.ident 获取线程的标识符。线程标识符是一个非零整数,只有在调用了start()方法之后该属性才有效，否则它只返回None。

　　Thread.is_alive() /Thread.isAlive 判断线程是否是激活的（alive）。从调用start()方法启动线程,到run()方法执行完毕或遇到未处理异常而中断这段时间内，线程是激活的。

　　Thread.join([timeout]) 调用Thread.join将会使线程堵塞，直到被调用线程运行结束或超时。参数timeout是一个数值类型，表示超时时间，如果未提供该参数，那么主调线程将一直堵塞到被调线程结束。

　　threading使用简介-3

　　Thread.join([timeout]) 调用Thread.join将会使主线程堵塞，直到被调用线程运行结束或超时。参数timeout是一个数值类型，表示超时时间，如果未提供该参数，那么主调线程将一直堵塞到被调线程结束。

　　什么是子线程？

　　　　包含在 threading.Thread中,里面均视为子线程。

　　什么是主线程？

　　　　除了“不包含在Thread里面的程序”,UI界面和Main函数均为主线程，均可视为主线程。

　　threading使用简介-4

　　Thread类还定义了以下常用方法与属性：

　　python对于thread的管理中有两个函数：

　　Join()和setDaemon()　　

　　join：如在一个线程B中调用thread1.join()，则thread1结束后，线程B才会接着threada.join()往后运行。 　　

　　setDaemon：主线程A启动了子线程B，调用B.setDaemaon(True)，则主线程结束时，会把子线程B也杀死。必须在运行线程之前设置.

　　Thread.isDaemon()/ Thread.daemon 用于获取线程的名称,有返回值,需要print打印出来

　　GIL全局解释锁简介

　　GIL并不是Python的特性，Python完全可以不依赖于GIL

　　为了更有效的利用多核处理器的性能，就出现了多线程的编程方式，而随之带来的就是线程间数据一致性和状态同步的困难。为了有效解决多份缓存之间的数据同步时各厂商花费了不少心思,就有了GIL,也不可避免的带来了一定的性能损失。

　　Python当然也逃不开，为了利用多核，Python开始支持多线程。而解决多线程之间数据完整性和状态同步的最简单方法自然就是加锁。 于是有了GIL这把超级大锁

　　GIL无疑就是一把全局排他锁。毫无疑问全局锁的存在会对多线程的效率有不小影响。甚至就几乎等于Python是个单线程的程序。 那么读者就会说了，全局锁只要释放的勤快效率也不会差啊。只要在进行耗时的IO操作的时候，能释放GIL，这样也还是可以提升运行效率的嘛。或者说再差也不会比单线程的效率差吧。理论上是这样，而实际上呢？Python比你想的更糟。

　　但当CPU有多个核心的时候，问题就来了。从release GIL到acquire GIL之间几乎是没有间隙的。所以当其他在其他核心上的线程被唤醒时，大部分情况下主线程已经又再一次获取到GIL了。这个时候被唤醒执行的线程只能白白的浪费CPU时间，看着另一个线程拿着GIL欢快的执行着。然后达到切换时间后进入待调度状态，再被唤醒，再等待，以此往复恶性循环。GIL的存在导致多线程无法很好的利用多核CPU的并发处理能力。

 

#多线程 python不鼓励使用多线程 python的多线程是伪多线程  效率不高
'''
def music(name,loop):
    for i in range(loop):
        # print('listen music %s %s %s'%(name,time.ctime(),threading.Thread.getNmae(t1)))
        print('listen music %s %s'%(name,time.ctime()))

        time.sleep(1)

def movie(name,loop):
    for i in range(loop):
        # print('listen movie %s %s %s'%(name,time.ctime(),threading.Thread.getNmae(t2)))
        print('listen movie %s %s'%(name,time.ctime()))
        time.sleep(1)

#创建多线程
t1 = threading.Thread(target=music,args=('爱的故事上集',3))
t1.setName('musicThread')
t2 = threading.Thread(target=movie,args=('肖生克的救赎',4),name='movieThread')

if __name__== '__main__':

    #守护主线程 如果主线程结束则杀死子线程
    t1.setDaemon(True)
    t2.setDaemon(True)


    #启动线程
    t1.start()
    t2.start()

    #线程ID
    # print(t1.ident)
    # print(t2.ident)

    #join可以对主线程进行阻塞 等子线程执行完在执行主线程
    # t1.join()
    # t2.join()
    print('主线程:%s'%time.ctime())

    # print('end time %s' %time.ctime())
'''

 

 

#加锁
balance = 0

def change(n):
    global balance
    balance += n
    balance -= n

# def run_thread(n):
#     for i in range(1000000):
#         change(n)

'''
lock = threading.Lock()#获取线程锁
def run_thread(n):
    for i in range(1000000):
        #加锁
        lock.acquire()
        try:
            change(n)
        finally:
            #释放锁
            lock.release()



t1 = threading.Thread(target= run_thread,args = (4,))
t2 = threading.Thread(target= run_thread,args = (8,))

t1.start()
t2.start()
t1.join()
t2.join()
print(balance)
'''

 

4.python多进程模块multiprocessing

　　用multiprocessing替代Thread

　　multiprocessing库的出现很大程度上是为了弥补thread库因为GIL低效的缺陷。它完整的复制了一套thread所提供的接口方便迁移。唯一的不同就是它使用了多进程而不是多线程。每个进程有自己的独立的GIL，完全并行，无GIL的限制(进程中包括线程)，可充分利用多cpu多核的环境,因此也不会出现进程之间的GIL争抢。

　　python多进程并发，模块名称：multiprocessing

　　python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。

　　借助这个包，可以轻松完成从单进程到并发执行的转换。

　　导入方式： import multiprocessing

　　Multiprocessing使用简介-1

　　multiprocessing包是Python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该Process对象与Thread对象的用法相同，也有start(), run(), join()等方法。

　　此外multiprocessing包中也有Lock/Event/Semaphore/Condition类 (这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的Thread类一致。所以，multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。

　　multiprocessing提供了threading包中没有的IPC(比如Pipe和Queue)，效率上更高。应优先考虑Pipe和Queue，避免使用Lock/Event/Semaphore/Condition等同步方式 (因为它们占据的不是用户进程的资源,而是线程)。

　　Process

　　创建进程的类：Process([group [, target [, name [, args [, kwargs]]]]])

　　方法：is_alive()、join([timeout])、run()、start() 。其中，Process以start()启动某个进程。

　　创建函数并作为进程

　　　　创建函数并将其作为单进程

　　　　创建函数并将其作为多个进程

　　　　将进程定义为类。

 

import time
import multiprocessing

#单进程
'''
def work_1(f,n):
    print('work_1_start')
    for i in range(n):
        with open(f,'a') as fs:
            fs.write('i love python n')
            time.sleep(1)
    print('work_1_end')


def work_2(f,n):
    print('work_2_start')
    for i in range(n):
        with open(f,'a') as fs:
            fs.write('come on baby n')
            time.sleep(1)
    print('work_2_end')

if __name__ == '__main__':
    work_1('file.txt',4)
    work_2('file.txt', 3)
'''

#多进程
'''
def work_1(f,n):
    print('work_1_start')
    for i in range(n):
        with open(f,'a') as fs:
            fs.write('i love python n')
            time.sleep(1)
    print('work_1_end')


def work_2(f,n):
    print('work_2_start')
    for i in range(n):
        with open(f,'a') as fs:
            fs.write('come on baby n')
            time.sleep(1)
    print('work_2_end')

if __name__ == '__main__':
    p1 = multiprocessing.Process(target=work_1,args = ('file.txt',3))
    p2 = multiprocessing.Process(target=work_2,args = ('file.txt',3))

    p1.start()
    p2.start()
'''

#加锁
def work_1(f,n,lock):
    print('work_1_start')
    lock.acquire()
    for i in range(n):
        with open(f,'a') as fs:
            fs.write('i love python n')
            time.sleep(1)
    print('work_1_end')
    lock.release()


def work_2(f,n,lock):
    print('work_2_start')
    lock.acquire()
    for i in range(n):
        with open(f,'a') as fs:
            fs.write('come on baby n')
            time.sleep(1)
    print('work_2_end')
    lock.release()

if __name__ == '__main__':
    lock = multiprocessing.Lock()
    p1 = multiprocessing.Process(target=work_1,args = ('file.txt',3,lock))
    p2 = multiprocessing.Process(target=work_2,args = ('file.txt',3,lock))

    p1.start()
    p2.start()

 

　　多进程的几种方法

　　Lock:可以避免访问资源时的冲突

　　Pool:可以提供指定数量的进程

　　Queue:多进程安全的队列，实现多进程之间的数据传递

　　Pipe:实现管道模式下的消息发送与接收

　　Lock

　　同步与异步

　　同步执行：一个进程在执行任务时，另一个进程必须等待执行完毕，才能继续执行,加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改.没错，速度是慢了，但牺牲了速度却保证了数据安全。

　　异步执行：一个进程在执行任务时，另一个进程无需等待其执行完毕就可以执行，当有消息返回时，系统会提醒后者进行处理，这样会很好的提高运行效率.

#加锁
def work_1(f,n,lock):
    print('work_1_start')
    lock.acquire()
    for i in range(n):
        with open(f,'a') as fs:
            fs.write('i love python n')
            time.sleep(1)
    print('work_1_end')
    lock.release()


def work_2(f,n,lock):
    print('work_2_start')
    lock.acquire()
    for i in range(n):
        with open(f,'a') as fs:
            fs.write('come on baby n')
            time.sleep(1)
    print('work_2_end')
    lock.release()

if __name__ == '__main__':
    lock = multiprocessing.Lock()
    p1 = multiprocessing.Process(target=work_1,args = ('file.txt',3,lock))
    p2 = multiprocessing.Process(target=work_2,args = ('file.txt',3,lock))

    p1.start()
    p2.start()

　　Pool

　　Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来执行它。

　　Pool方法

　　进程池方法：

　　apply(func[, args[, kwds]])： 阻塞的执行，比如创建一个有3个线程的线程池，当执行时是创建完一个,执行完函数再创建另一个，变成一个线性的执行.

　　apply_async(func[, args[, kwds[, callback]]]) ： 它是非阻塞执行，同时创建3个线程的线城池，同时执行，只要有一个执行完立刻放回池子待下一个执行，并行的执行 .

　　close()： 关闭pool，使其不在接受新的任务。

　　terminate() ： 结束工作进程，不在处理未完成的任务。

　　join(): 主进程阻塞，等待子进程的退出， join方法要在close或terminate之后使用。

 

import os
import time
import multiprocessing

# def work(n):
#     print('run work (%s),work id %s' %(n,os.getpid()))
#     time.sleep(3)
#     print('work (%s) stop,work id %s' %(n,os.getpid()))
#
# if __name__=='__main__':
#     print('Parent process %s.' % os.getpid())
#     #创建进程池
#     p = multiprocessing.Pool(3)
#     for i in range(5):
#         #创建5个进程，依次进入进程池
#         p.apply_async(work,args = (i,))
#     p.close()
#     p.join()


def music(name,loop):
    print(time.ctime())
    for i in range(loop):
        time.sleep(2)
        print('您现在正在听得音乐是%S'%name)

def movie(name,loop):
    print(time.ctime())
    for i in range(loop):
        time.sleep(2)
        print('您现在正在看的电影是%s'%name)

if __name__=='__main__':
    pool = multiprocessing.Pool(2)
    pool.apply_async(func=music,args=('花太香',3))
    pool.apply_async(func=movie, args=('王牌特工', 4))
    pool.apply_async(func=music, args=('爱的故事上集', 2))
    pool.close()
    # pool.terminate()
    #比较危险，不要轻易使用，直接杀死进程池
    pool.join()#join阻塞主进程，当子进程执行完毕后继续往后执行，
    # 使用join必须在进程池中使用close或者terminate，join要在其之后使用
    print('结束时间是%s'%time.ctime())

 

　　Queue

　　进程彼此之间互相隔离，要实现进程间通信（IPC）,multiprocessing模块支持两种形式：队列Queue和管道Pipe，这两种方式都是使用消息传递的

　　创建队列的类（底层就是以管道和锁定的方式实现）：

　　Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。      

　　参数介绍：maxsize是队列中允许最大项数，省略则无大小限制。

　　Queue方法

　　q.put方法用以插入数据到队列中

　　put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。

　　q.get方法可以从队列读取并且删除一个元素

　　同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.

　　q.get_nowait():同q.get(False)

　　q.put_nowait():同q.put(False)

　　q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。

　　q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。

　　q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同上.

 

import multiprocessing
import time

#跨进程通信
def put(q):
    for value in ['A', 'B', 'C']:
        print('发送%s到queue...'%value)
        q.put(value) #通过put方法发送
        time.sleep(2)

#读数据进程执行的代码
def get(q):
    while True:
        value = q.get(True)#接受队列中的数据
        print('从queue接受 %s.'% value)

if __name__ == '__main__':
    #父进程创建queue，并传递给各个子进程
    q = multiprocessing.Queue()

    pw = multiprocessing.Process(target=put, args = (q,))
    pr = multiprocessing.Process(target=get, args = (q,))
    #启动子进程pw 写入
    pw.start()
    #启动子进程pr 读取
    pr.start()
    #等待pw结束
    pw.join()
    #pr进程里面是死循环，无法等待其结束，只能强行终止
    pr.terminate()

 

　　Pipe

　　Pipe方法返回(conn1, conn2)代表一个管道的两个端。Pipe方法有duplex参数:duplex 为 True(默认值)，那么这个管道是全双工模式，也就是说conn1和conn2均可收发。duplex 为 False，conn1只负责接受消息，conn2只负责发送消息。

　　send和recv方法分别是发送和接收消息的方法。在全双工模式下，可以调用conn1.send发送消息，conn1.recv接收消息。如果没有消息可接收，recv方法会一直阻塞。如果管道已经被关闭，那么recv方法会抛出EOFError

 

import multiprocessing
import time

#pipe管道
def put(p):
    for value in ['A','B','C']:
        print('发送 %s 到 pipi...' % value)
        p[1].send(value)
        time.sleep(2)

#读数据进程执行的代码
def get(p):
    while True:
        value = p[0].recv()
        print('从 pipe 接受 %s' %value)

if __name__ == '__main__':
    #父进程创建pipe 并传递给各个子进程
    # p = multiprocessing.Pipe()
    p = multiprocessing.Pipe(duplex=False)  # 左收右发
    pw = multiprocessing.Process(target= put,args=(p,))
    pr = multiprocessing.Process(target=get, args=(p,))
    # 启动子进程pw 写入
    pw.start()
    # 启动子进程pr 读取
    pr.start()
    # 等待pw结束
    pw.join()
    # pr进程里面是死循环，无法等待其结束，只能强行终止
    pr.terminate()

 

　　Python—队列、生产者消费者模型

　　queue.Queue是进程内非阻塞队列

　　multiprocess.Queue是跨进程通信队列。

　　队列：FIFO(先进先出)、

　　LIFO(后进先出) 生产者消费者模型

 

import threading
import time
import queue

#生产者消费者模型

q = queue.Queue(maxsize=10)
def producer(name):  # 生产者
    count = 1
    while True:
        q.put("骨头%s" % count)
        print("生产了骨头", count)
        count += 1
        time.sleep(0.5)


def consumer(name):  # 消费者
    while True:
        print("[%s]取到[%s]并且吃了它..." % (name, q.get()))
        time.sleep(1)


p = threading.Thread(target=producer, args=("Tim",))
c1 = threading.Thread(target=consumer, args=("King",))
c2 = threading.Thread(target=consumer, args=("Wang",))

p.start()
c1.start()
c2.start()