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

2023-07-18 15:49:28

一、线程

　　多任务可以由多进程完成，也可以由一个进程内的多线程完成，一个进程内的所有线程，共享同一块内存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方法，将线程对象添加到队列中去，而且类初始化的时候，一次性将所有的线程类都添加到队列中去了，总之上面的线程池虽然实现简单，但是实际上却有很多问题，下面来看一个真正意义上的线程池。

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

队列中存的不是线程类，而是任务，我们从队列中拿取的都是任务
每次执行任务的时候，不是都要生成一个线程，而是如果以前生成的线程有空闲的话，就用以前的线程
支持回掉机制，支持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，以后会继续更新的，

码农公寓

一、线程

二、线程锁

三、线程间通信Event

四、队列

五、进程

六、进程间数据共享

1、Value，Array

2、manage

七、进程池

八、线程池

九、协程

十、小结

相关文章

四、队列　　

五、进程　

六、进程间数据共享　　

2、manage　　

七、进程池　　

八、线程池　

九、协程　

十、小结　