Python全栈开发之11进程和线程

Posted 2020-07-28 1000-7

一、线程

　　多任务可以由多进程完成，也可以由一个进程内的多线程完成，一个进程内的所有线程，共享同一块内存python中创建线程比较简单，导入threading模块，下面来看一下代码中如何创建多线程。

def f1(i):
    time.sleep(1)
    print(i)

if __name__ == \'__main__\':
    for i in range(5):
        t = threading.Thread(target=f1, args=(i,))
        t.start()
    print(\'start\')          # 主线程等待子线程完成，子线程并发执行


>>start
>>2
>>1
>>3
>>0
>>4

　　主线程从上到下执行，创建5个子线程，打印出\'start\'，然后等待子线程执行完结束，如果想让线程要一个个依次执行完，而不是并发操作，那么就要使用join方法。下面来看一下代码

import threading
import time

def f1(i):
    time.sleep(1)
    print(i)

if __name__ == \'__main__\':
    for i in range(5):
        t = threading.Thread(target=f1, args=(i,))
        t.start()
        t.join()
    print(\'start\')      # 线程从上到下依次执行，最后打印出start

>>0
>>1
>>2
>>3
>>4
>>start

　　上面的代码不适用join的话，主线程会默认等待子线程结束，才会结束，如果不想让主线程等待子线程的话，可以子线程启动之前设置将其设置为后台线程，如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止，前台线程则相反，若果不加指定的话，默认为前台线程，下面从代码来看一下，如何设置为后台线程。例如下面的例子，主线程直接打印start，执行完后就结束，而不会去等待子线程，子线程中的数据也就不会打印出来

import threading
import time

def f1(i):
    time.sleep(1)
    print(i)

if __name__ == \'__main__\':
    for i in range(5):
        t = threading.Thread(target=f1, args=(i,))
        t.setDaemon(True)
        t.start()

    print(\'start\')      # 主线程不等待子线程

>> start　

　　除此之外，自己还可以为线程自定义名字，通过 t = threading.Thread(target=f1, args=(i,), name=\'mythread{}\'.format(i)) 中的name参数，除此之外，Thread还有一下一些方法

• t.getName() : 获取线程的名称
• t.setName() ： 设置线程的名称 
• t.name : 获取或设置线程的名称
• t.is_alive() ： 判断线程是否为激活状态
• t.isAlive() ：判断线程是否为激活状态
• t.isDaemon() ： 判断是否为守护线程

二、线程锁

　　由于线程是共享同一份内存的，所以如果操作同一份数据，很容易造成冲突，这时候就可以为线程加上一个锁了，这里我们使用Rlock，而不使用Lock，因为Lock如果多次获取锁的时候会出错，而RLock允许在同一线程中被多次acquire，但是需要用n次的release才能真正释放所占用的琐，一个线程获取了锁在释放之前，其他线程只有等待。　

import threading
G = 1
lock = threading.RLock()
def fun():
    lock.acquire()    # 获取锁
    global G
    G += 2
    print(G, threading.current_thread().name)
    lock.release()   # 释放锁
    return


for i in range(10):
    t = threading.Thread(target=fun, name=\'t-{}\'.format(i))
    t.start()

3 t-0
5 t-1
7 t-2
9 t-3
11 t-4
13 t-5
15 t-6
17 t-7
19 t-8
21 t-9    

三、线程间通信Event

Event是线程间通信最间的机制之一，主要用于主线程控制其他线程的执行，主要用过wait，clear，set，这三个方法来实现的的，下面来看一个简单的例子，

import threading
import time

def f1(event):
    print(\'start:\')
    event.wait()            # 阻塞在，等待 set
    print(\'end:\')

if __name__ == \'__main__\':
    event_obj  = threading.Event()
    for i in range(5):
        t = threading.Thread(target=f1, args=(event_obj,))
        t.start()

    event_obj.clear()      # 清除标志位 
    inp = input(\'>>>>:\')
    if inp == \'true\':
        event_obj.set()   # 设置标志位

四、队列　　

　　可以简单的理解为一种先进先出的数据结构，比如用于生产者消费者模型，或者用于写线程池，以及前面写select的时候，读写分离时候可用队列存储数据等等，以后用到队列的地方很多，因此对于队列的用法要熟练掌握。下面首先来看一下队列提供了哪些用法

q = queue.Queue(maxsize=0)  # 构造一个先进显出队列，maxsize指定队列长度，为0时，表示队列长度无限制。

q.join()    　　　　# 等到队列为kong的时候，在执行别的操作
q.qsize()   　　　　# 返回队列的大小 （不可靠）
q.empty()   　　　　# 当队列为空的时候，返回True 否则返回False （不可靠）
q.full()    　　　　# 当队列满的时候，返回True，否则返回False （不可靠）
q.put(item, block=True, timeout=None) 　　# 将item放入Queue尾部，item必须存在，参数block默认为True,表示当队列满时，会等待
　　　　　　　　　　　　　　　　　　　　　　　　# 为False时为非阻塞，此时如果队列已满，会引发queue.Full 异常。 可选参数timeout，表示会阻塞设置的时间，
　　　　　　　　　　　　　　　　　　　　　　　　# 如果在阻塞时间里 队列还是无法放入，则引发 queue.Full 异常

q.get(block=True, timeout=None) 　　　　#  移除并返回队列头部的一个值，可选参数block默认为True，表示获取值的时候，如果队列为空，则阻塞
　　　　　　　　　　　　　　　　　　　　　　 #  阻塞的话若此时队列为空，则引发queue.Empty异常。 可选参数timeout，表示会阻塞设置的时间，
q.get_nowait() 　　　　　　　　　　　　　　#  等效于 get(item,block=False)　

下面用代码来简单的演示下，消费者生成者模型，只是简单的演示下。

message = queue.Queue(10)

def product(num):
    for i in range(num):
        message.put(i)
        print(\'将{}添加到队列中\'.format(i))
        time.sleep(random.randrange(0, 1))


def consume(num):
    count = 0
    while count<num:
        i = message.get()
        print(\'将{}从队列取出\'.format(i))
        time.sleep(random.randrange(1, 2))
        count += 1


t1 = threading.Thread(target=product, args=(10, ))
t1.start()

t2 = threading.Thread(target=consume, args=(10, ))
t2.start()

五、进程　

　　线程的上一级就是进程，进程可包含很多线程，进程和线程的区别是进程间的数据不共享，多进程也可以用来处理多任务，不过多进程很消耗资源，计算型的任务最好交给多进程来处理，IO密集型最好交给多线程来处理，此外进程的数量应该和cpu的核心说保持一致。

在windows中不能用fork来创建多进程，因此只能导入multiprocessing，来模拟多进程，下面首先来看一下怎么创建进程，大家可以先猜一下下面的结果是什么

l = []

def f(i):
    l.append(i)
    print(\'hi\', l)

if __name__ == \'__main__\':
    for i in range(10):
        p = multiprocessing.Process(target=f, args=(i,))        # 数据不共享，创建10份 l列表
        p.start()

六、进程间数据共享　　

进程间的数据是不共享的，但是我如果非要数据共享了，那么就需要用其他方式了 

1、Value，Array

def f(a, b):
    a.value = 3.111
    for i in range(len(b)):
        b[i] += 100

if __name__ == \'__main__\':
    num = Value(\'f\', 3.333)　　　　　　　　# 类似C语言中的 浮点型数
    l = Array(\'i\', range(10))　　　　　　 # 类似C语言中的整形数组，长度为10
    print(num.value)
    print(l[:])

    p = Process(target=f, args=(num, l))
    p.start()
    p.join()
    print(num.value)　　　　　　　　　　　　　　# 大家自己运行一下，看下两次打印结果是否一样
    print(l[:])

2、manage　　

方式一，使用的都是C语言中的数据结构，如果大家对c不熟悉的话，用起来比较麻烦，方式2就可以支持python自带的数据，下面来看一下

from multiprocessing import Process,Manager

def Foo(dic, i):
    dic[i] = 100 + i
    print(dic.values())

if __name__ == \'__main__\':
    manage = Manager()
    dic = manage.dict()

    for i in range(2):
        p = Process(target=Foo, args=(dic, i))
        p.start()
        p.join()

七、进程池　　

　　实际应用中，并不是每次执行任务的时候，都去创建多进程，而是维护了一个进程池，每次执行的时候，都去进程池取一个，如果进程池里面的进程取光了，就会阻塞在那里，直到进程池中有可用进程为止。首先来看一下进程池提供了哪些方法

• apply(func[, args[, kwds]]) ：使用arg和kwds参数调用func函数，结果返回前会一直阻塞，由于这个原因，apply_async()更适合并发执行，另外，func函数仅被pool中的一个进程运行。

• apply_async(func[, args[, kwds[, callback[, error_callback]]]]) ： apply()方法的一个变体，会返回一个结果对象。如果callback被指定，那么callback可以接收一个参数然后被调用，当结果准备好回调时会调用callback，调用失败时，则用error_callback替换callback。 Callbacks应被立即完成，否则处理结果的线程会被阻塞。

• close() ： 等待任务完成后在停止工作进程，阻止更多的任务提交到pool，待任务完成后，工作进程会退出。

• terminate() ： 不管任务是否完成，立即停止工作进程。在对pool对象进程垃圾回收的时候，会立即调用terminate()。

• join() : 等待工作线程的退出，在调用join()前，必须调用close() or terminate()。这样是因为被终止的进程需要被父进程调用wait（join等价与wait，否则进程会成为僵尸进程。

下面来简单的看一下代码怎么用的

from multiprocessing import Pool
import time

def f1(i):
    time.sleep(1)
    # print(i)
    return i

def cb(i):
    print(i)

if __name__ == \'__main__\':
    poo = Pool(5)
    for i in range(20):
        # poo.apply(func=f1, args=(i,))   # 串行执行，排队执行 有join
        poo.apply_async(func=f1, args=(i,), callback=cb)  # 并发执行 主进程不等子进程，无join
    print(\'**********\')

    poo.close()
    poo.join()

八、线程池　

　　对于前面的进程池，python自带了一个模块Pool供我们使用，但是对于线程池，则没有提供，因此需要我们自己写，自己写的话，就需要用到队列，下面我们来看一下自己怎么实现一个线程池，首先写一个最简单的版本。　

import threading
import time
import queue

class ThreadPool:
    def __init__(self, max_num=20):
        self.queue = queue.Queue(max_num)
        for i in range(max_num):
            self.add()

    def add(self):
        self.queue.put(threading.Thread)

    def get(self):
        return self.queue.get()

def f(tp, i):
    time.sleep(1)
    print(i)
    tp.add()

p = ThreadPool(10)
for i in range(20):
    thread = p.get()
    t = thread(target=f, args=(p, i))
    t.start()

上述代码写了一个线程池类，基本实现了线程池的功能，但是有很多缺点，没有实现回掉函数，每次执行任务的时候，任务处理函数每次执行完都需要自动执行对象的add方法，将线程对象添加到队列中去，而且类初始化的时候，一次性将所有的线程类都添加到队列中去了，总之上面的线程池虽然实现简单，但是实际上却有很多问题，下面来看一个真正意义上的线程池。

　　在写代码之前，我们先来看一下该怎么设计这样一个线程池，上面的线程池，我们的队列中，存的是线程类，我们每处理一个任务都实例化一个线程，然后执行完了之后，该线程就被丢弃了，这样有点不合适。我们这次设计的时候，

1. 队列中存的不是线程类，而是任务，我们从队列中拿取的都是任务
2. 每次执行任务的时候，不是都要生成一个线程，而是如果以前生成的线程有空闲的话，就用以前的线程
3. 支持回掉机制，支持close，terminate

下面来一下代码是怎么实现的

import threading
import queue
import time
import contextlib

class ThreadingPool:
    def __init__(self, num):
        self.max = num
        self.terminal = False
        self.q = queue.Queue()
        self.generate_list = []         # 保存已经生成的线程
        self.free_list = []             # 保存那些已经完成任务的线程

    def run(self, func, args=None, callbk=None):
        self.q.put((func, args, callbk))            # 将任务信息作为一个元祖放到队列中去
        if len(self.free_list) == 0 and len(self.generate_list) < self.max:
           self.threadstart()

    def threadstart(self):
        t = threading.Thread(target=self.handel)
        t.start()

    def handel(self):
        current_thread = threading.current_thread()
        self.generate_list.append(current_thread)
        event = self.q.get()
        while event != \'stop\':
            func, args, callbk = event
            flag = True
            try:
                ret = func(*args)
            except Exception as e:
                flag = False
                ret = e

            if callbk is not None:
                try:
                    callbk(ret)
                except Exception as e:
                    pass

            if not self.terminal:
                with self.auto_append_remove(current_thread):
                    event = self.q.get()
            else:
                event = \'stop\'
        else:
            self.generate_list.remove(current_thread)

    def terminate(self):
        self.terminal = True

        while self.generate_list:
            self.q.put(\'stop\')
        self.q.empty()

    def close(self):
        num = len(self.generate_list)
        while num:
            self.q.put(\'stop\')
            num -= 1

    @contextlib.contextmanager
    def auto_append_remove(self, thread):
        self.free_list.append(thread)
        try:
            yield
        finally:
            self.free_list.remove(thread)

def f(i):
    # time.sleep(1)
    return i

def f1(i):
    print(i)

p = ThreadingPool(5)
for i in range(20):
    p.run(func=f, args=(i,), callbk=f1)

p.close()

九、协程　

协程，又称微线程，协程执行看起来有点像多线程，但是事实上协程就是只有一个线程，因此，没有线程切换的开销，和多线程比，线程数量越多，协程的性能优势就越明显，此外因为只有一个线程，不需要多线程的锁机制，也不存在同时写变量冲突。协程的适用场景：当程序中存在大量不需要CPU的操作时（IO）下面来看一个利用协程例子

from gevent import monkey
import gevent
import requests

# 把标准库中的thread/socket等给替换掉
# 这样我们在后面使用socket的时候可以跟平常一样使用,无需修改任何代码,但是它变成非阻塞的了.
monkey.patch_all()　　　　　　# 猴子补丁

def f(url):
    print(\'GET: %s\' % url)
    resp = requests.get(url)
    data = resp.text
    print(\'%d bytes received from %s.\' % (len(data), url))

gevent.joinall([
        gevent.spawn(f, \'https://www.python.org/\'),
        gevent.spawn(f, \'https://www.yahoo.com/\'),
        gevent.spawn(f, \'https://github.com/\'),
])

上面的例子，利用协程，一个线程完成所有的请求，发出请求的时候，不会等待回复，而是一次性将所有的请求都发出求，收到一个回复就处理一个回复，这样一个线程就解决了所有的事情，效率极高。

十、小结　

这篇博文是pyton基础知识的最后一篇，后面会讲的博文会讲开始讲前端的知识，这里附上目录http://www.cnblogs.com/Wxtrkbc/p/5606048.html，以后会继续更新的，