[阶段一] 15. 进程与线程编程

末蓝、 2022-09-12 11:45 172阅读 0赞

### 进程与线程编程 ###

*  进程与多进程：

进程就是程序执行的载体。打开的每个软件、游戏，执行的每一个 python 脚本，都是启动一个进程。对于系统来说，一个任务就是一个进程，多任务就是系统在同时运行多个进程。

就像人需要吃饭一样，进程的口粮是 cpu 和内存资源，不同进程需要的 cpu 和内存资源不同。当一个任务被开启后，操作系统会分配它所需的系统资源，包括内存、I/O 和 cpu 等，如果系统资源不够，则会出现系统崩溃的情况，这样的任务可被称为进程。进程之间不能共享内存。

对于系统来说，如果每次只能启动一个进程，那这个系统就是单进程系统。多进程就是同时启动多个进程，进程之间互不干扰，各自执行自己的业务逻辑。现在的系统基本上都是多进程系统，启动新的进程不再需要关闭之前的进程。

以手机系统为例，系统进程可以看做是父进程，而QQ、微信、王者等软件就是一个个子进程，子进程下面还可以有各自的子进程。

*  线程与多线程：

线程是操作系统最小的执行单元，进程至少由一个线程组成。如何调度进程和线程，完全由操作系统决定，程序自身无法决定。

可以这样理解，在一个进程内部，要同时干多件事，就需要同时运行多个“子任务”，进程内的这些“子任务”就是线程。

线程依赖于进程，进程提供线程执行程序的前置要求，线程在重组的资源配置下，去执行程序。而一个进程有多个线程就涉及到进程有多少可以被 cpu 单独调用的模块，这个调用的模块可以通过手动创建线程来建立。

一个线程可以创建和撤销另一个线程，线程也可以启动多个线程，同一进程中的多个线程之间可以并发执行，多个线程共享进程的 cpu 和内存等系统资源。此外，与进程类似，线程下面还可以有各自的子线程。

*  进程与线程的区别：

进程可以被称为执行的程序，一个进程拥有完整的数据空间和代码空间，每一个进程的地址空间都是独立的，进程之间不共享数据。

进程是进程的一个实体，是 CPU 调度和分派的基本单位，它是比进程更小的能独立运行的基本单位。线程自己基本上不拥有系统资源，只拥有一点在运行中必不可少的资源，但是它可与同属一个进程的其它线程共享进程所拥有的全部资源。

二者之间的关系：

一个程序至少有一个进程，一个进程下至少有一个线程

区别：

1. 进程有独立的地址空间，多进程较稳定，因为其中一个出现状况不影响另外一个；同一个进程的多个线程，共用地址空间，多线程相比于多进程，稳定性更差，因为一个线程出现问题会严重影响其它线程。
    
    2. 进程之间需要共享数据，要利用进程间通讯来实现；同一个进程中的线程不需要
    
    3. 进程只是资源分配的最小单位；线程是执行的最小单位，也就是实际执行的是线程

CPU 密集型和 IO 密集型：

1. CPU 密集型代码（各种循环处理、计数等）
    
    2. IO 密集型代码（文件处理、网络爬虫等）

因为 python 锁的问题，线程会进行锁竞争、切换线程，这就会消耗资源。在 CPU 密集型任务下，多进程更快或者说效果更好；而对于 IO 密集型任务，多线程能有效提高效率。

*  多进程的创建：

在 python 中创建进程：

使用的模块：multiprocessing
    
    使用的类：multiprocessing.Process

`multiprocessing.Process` 的功能是创建一个进程。

用法：

import multiprocessing
    
    multiprocessing.Process(target, args)

参数 `target` 表示一个函数名，`args` 表示对应 `target` 函数的参数。返回一个进程对象。

`multiprocessing.Process` 类的常用方法：

<table> 
 <thead> 
  <tr> 
   <th>函数名</th> 
   <th>介绍</th> 
   <th>参数</th> 
   <th>返回值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>start</td> 
   <td>执行进程</td> 
   <td>无</td> 
   <td>无</td> 
  </tr> 
  <tr> 
   <td>join</td> 
   <td>阻塞进程</td> 
   <td>无</td> 
   <td>无</td> 
  </tr> 
  <tr> 
   <td>kill</td> 
   <td>杀死进程</td> 
   <td>无</td> 
   <td>无</td> 
  </tr> 
  <tr> 
   <td>is_alive</td> 
   <td>进程是否存活</td> 
   <td>无</td> 
   <td>bool</td> 
  </tr> 
 </tbody> 
</table>

示例：

# coding:utf-8
    
    import time
    import os
    import multiprocessing
    
    
    def work_a():
        for i in range(10):
            print(i, 'a', os.getpid())
            time.sleep(1)
    
    
    def work_b():
        for i in range(10):
            print(i, 'b', os.getpid())
            time.sleep(1)
    
    
    if __name__ == '__main__':
        start = time.time()
        process_a = multiprocessing.Process(target=work_a)
        process_a.start()
        # process_a.join()
    
        process_b = multiprocessing.Process(target=work_b)
        process_b.start()
    
        print(time.time() - start)
        print('parent pid is %s' % (os.getpid()))

多进程可以提高运行的效率，但也会带来一些问题。多进程的问题：

通过进程模块执行的函数无法获取返回值，即使函数内有 return 关键字 —— 通过队列解决
    
    多个进程同时修改文件可能会出现错误 —— 创建进程锁解决
    
    进程数量太多可能会造成资源不足，甚至死机等情况 —— 创建进程池解决

*  进程池与进程锁：

进程池可以理解为一个进程的池子，这个池子中已经提前创建好了一定数量的进程。进程池中的进程创建好之后，会被重复使用而不会关闭，因此也就避免了创建与关闭的消耗。进程池关闭时，内部的进程也会随之关闭。

当有任务需要执行时，首先会判断进程池中有没有空闲的进程，如果有空闲进程则任务会选择其中一个执行任务，没有空闲进程则任务会等待，直到进程池中有空闲进程。

在 python 中创建进程池：

使用的模块：multiprocessing
    
    使用的类：multiprocessing.Pool

`multiprocessing.Pool` 的功能是创建一个进程池。

用法：

import multiprocessing
    
    multiprocessing.Pool(Processcount)

参数 `Processcount` 表示进程池中的进程数量，返回一个进程池对象。

`multiprocessing.Pool` 类的常用方法：

<table> 
 <thead> 
  <tr> 
   <th>函数名</th> 
   <th>介绍</th> 
   <th>参数</th> 
   <th>返回值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>apply_async</td> 
   <td>任务加入进程池（异步）</td> 
   <td>func, args</td> 
   <td>无</td> 
  </tr> 
  <tr> 
   <td>close</td> 
   <td>关闭进程池</td> 
   <td>无</td> 
   <td>无</td> 
  </tr> 
  <tr> 
   <td>join</td> 
   <td>等待进程池任务结束</td> 
   <td>无</td> 
   <td>无</td> 
  </tr> 
 </tbody> 
</table>

示例：

# coding:utf-8
    
    import os
    import time
    import multiprocessing
    
    
    def work(count):
        print(count, os.getpid())
        time.sleep(1)
    
    
    if __name__ == '__main__':
        pool = multiprocessing.Pool(5)
    
        for i in range(20):
            pool.apply_async(func=work, args=(i,))
    
        pool.close()
        pool.join()

进程锁是仅针对于进程的锁，与之对应的还有线程锁。

当一个任务执行时，它会给进程加上进程锁，这样其它的任务就无法使用当前正在执行任务的进程。而当任务被执行完成时，进程锁会被释放，进程重新恢复为空闲状态，这样其它的任务就可以使用这个进程了。

可以看到，进程锁具有独占性，在同一时刻，一个进程只能被一个任务独占。整个加上、释放进程锁的过程是循环往复的。

在 python 中创建进程锁：

使用的模块：multiprocessing
    
    使用的类：multiprocessing.Manager

`multiprocessing.Manager` 的功能是可以对进程锁进行加锁与解锁。

用法：

import multiprocessing
    
    manager = multiprocessing.Manager()
    lock = manager.Lock()
    
    lock.acquire()
    lock.release()

`multiprocessing.Manager.Lock` 类的常用方法：

<table> 
 <thead> 
  <tr> 
   <th>函数名</th> 
   <th>介绍</th> 
   <th>参数</th> 
   <th>返回值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>acquire</td> 
   <td>加锁</td> 
   <td>无</td> 
   <td>无</td> 
  </tr> 
  <tr> 
   <td>release</td> 
   <td>解锁</td> 
   <td>无</td> 
   <td>无</td> 
  </tr> 
 </tbody> 
</table>

示例：

# coding:utf-8
    
    import os
    import time
    import multiprocessing
    
    
    def work(count, lock):
        lock.acquire()              #加锁
    
        print(count, os.getpid())
        time.sleep(1)
    
        lock.release()              #解锁
    
    
    if __name__ == '__main__':
        pool = multiprocessing.Pool(5)
        manager = multiprocessing.Manager()
        lock = manager.Lock()
    
        for i in range(20):
            pool.apply_async(func=work, args=(i, lock))
    
        pool.close()
        pool.join()

使用进程锁时需要注意，应该避免产生死锁，死锁时会导致后续进程无法继续执行任务，程序停止运行。

*  进程间的通信：

进程间的通信需要队列的支持。对列是一种数据存储结构，它的数据存储特点类似于排队，先进入队列的会先出来，后进入队列的后出来，因此它的数据只要通过 `put` 方法放入、`get` 方法取出即可。不需要安排取哪些数据进程的数据可放入队列，哪些进程需要，从队列中取出即可使用。

python 提供了多种进程通信的方式，主要是 `Queue` 和 `Pipe` 这两个类，`Queue` 用于多个进程间实现通信，`Pipe` 是两个进程的通信。其中 `Queue` 主要有两种方法即 `put` 和 `get` 方法，`put` 方法主要是以插入数据到队列中，`get` 方法是从队列读取并且删除一个元素。

在 python 中创建队列：

使用的模块：multiprocessing
    
    使用的类：multiprocessing.Queue

`multiprocessing.Queue` 的功能是创建一个队列。

用法：

import multiprocessing
    
    multiprocessing.Queue(mac_count)

参数 `mac_count` 表示队列可以传入的最大消息长度，默认为无限长度。返回一个队列对象。

`multiprocessing.Queue` 类的常用方法：

<table> 
 <thead> 
  <tr> 
   <th>函数名</th> 
   <th>介绍</th> 
   <th>参数</th> 
   <th>返回值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>put</td> 
   <td>信息放入队列</td> 
   <td>message</td> 
   <td>无</td> 
  </tr> 
  <tr> 
   <td>get</td> 
   <td>获取队列信息</td> 
   <td>无</td> 
   <td>str</td> 
  </tr> 
 </tbody> 
</table>

示例：

# coding:utf-8
    
    import time
    import json
    import multiprocessing
    
    
    class Work(object):
        def __init__(self, queue):
            self.queue = queue
    
        def send(self, message):
            if not isinstance(message, str):
                message = json.dumps(message)
            self.queue.put(message)
    
        def send_all(self):
            for i in range(10):
                self.queue.put(i)
                time.sleep(1)
    
        def receive(self):
            while True:
                result = self.queue.get()
    
                try:
                    res = json.loads(result)
                except:
                    res = result
    
                print('receive: %s' % res)
    
    
    if __name__ == '__main__':
        q = multiprocessing.Queue()
        work = Work(q)
    
        send = multiprocessing.Process(target=work.send, args=({ 'name': 'xiaobai'},))
        send_all = multiprocessing.Process(target=work.send_all)
        receive = multiprocessing.Process(target=work.receive)
    
        send.start()
        send_all.start()
        receive.start()
    
        send_all.join()
        receive.terminate()

*  线程的创建：

在 python 中创建线程：

使用的模块：threading
    
    使用的类：threading.Thread

`threading.Thread` 的功能是创建一个线程。

用法：

import multiprocessing
    
    multiprocessing.Thread(target, args)

参数 `target` 表示一个函数名，`args` 表示对应 `target` 函数的参数。返回一个线程对象。

`threading.Thread` 类的常用方法：

<table> 
 <thead> 
  <tr> 
   <th>函数名</th> 
   <th>说明</th> 
   <th>方法</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>start</td> 
   <td>启动进程</td> 
   <td>start()</td> 
  </tr> 
  <tr> 
   <td>join</td> 
   <td>阻塞直到线程执行结束</td> 
   <td>join(timeout=None)</td> 
  </tr> 
  <tr> 
   <td>getName</td> 
   <td>获取线程名</td> 
   <td>getName()</td> 
  </tr> 
  <tr> 
   <td>setName</td> 
   <td>设置线程名</td> 
   <td>setName()</td> 
  </tr> 
  <tr> 
   <td>is_alive</td> 
   <td>判断线程是否存活</td> 
   <td>is_alive()</td> 
  </tr> 
  <tr> 
   <td>setDaemon</td> 
   <td>守护线程</td> 
   <td>setDaemon(True)</td> 
  </tr> 
 </tbody> 
</table>

示例：

# coding:utf-8
    
    import time
    import random
    import threading
    
    lists = ['python', 'django', 'flask', 'tornado', 'bs5', 'requests', 'uvloop']
    new_lists = []
    
    
    def work():
        if len(lists) == 0:
            return
    
        data = random.choice(lists)
        lists.remove(data)
        new_data = '%s_new' % data
        new_lists.append(new_data)
        time.sleep(1)
    
    
    if __name__ == '__main__':
        start = time.time()
        t_list = []
        for i in range(len(lists)):
            t = threading.Thread(target=work)
            t_list.append(t)
            t.start()
    
        for t in t_list:
            t.join()
    
        print('old list：', lists)
        print('new list：', new_lists)
        print('time：', time.time() - start)

线程与进程的使用几乎一致，能大大提高程序的运行效率，但也会带来一些问题。线程的问题：

通过线程执行的函数无法获取返回值 —— 通过队列解决
    
    多个线程同时修改文件可能造成数据错乱 —— 创建线程锁解决
    
    线程数量太多可能会造成资源不足，甚至死机等情况 —— 创建线程池

*  线程池：

在 python 中创建线程池：

使用的模块：concurrent
    
    使用的类：concurrent.futures.ThreadPoolExecutor

`concurrent.futures.ThreadPoolExecutor` 的功能是创建一个线程池。

用法：

import concurrent
    
    concurrent.futures.ThreadPoolExecutor(max_workers)

参数 `max_workers` 表示线程池中的线程数量，返回一个线程池对象。

`concurrent.futures.ThreadPoolExecutor` 类的常用方法：

<table> 
 <thead> 
  <tr> 
   <th>函数名</th> 
   <th>介绍</th> 
   <th>参数</th> 
   <th>返回值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>submit</td> 
   <td>往线程池中加入任务</td> 
   <td>target, args</td> 
   <td>无</td> 
  </tr> 
  <tr> 
   <td>done</td> 
   <td>线程池中的某个线程是否完成了任务</td> 
   <td>无</td> 
   <td>bool</td> 
  </tr> 
  <tr> 
   <td>result</td> 
   <td>获取当前线程执行任务的结果</td> 
   <td>无</td> 
   <td>执行结果</td> 
  </tr> 
 </tbody> 
</table>

线程锁也具有独占性，在同一时刻，一个线程只能被一个任务独占。整个加上、释放线程锁的过程是循环往复的。

在 python 中创建线程锁：

使用的模块：threading
    
    使用的类：threading.Lock

`threading.Lock` 的功能是可以对线程锁进行加锁与解锁。

用法：

import threading
    
    lock = threading.Lock()
    
    lock.acquire()
    lock.release()

`threading.Lock` 类的常用方法：

示例：

# coding:utf-8
    
    import os
    import time
    import threading
    from concurrent.futures import ThreadPoolExecutor
    
    lock = threading.Lock()
    
    
    def work(count):
        lock.acquire()              #加锁
        time.sleep(1)
        lock.release()              #解锁
        return 'result：%s %s' % (count, os.getpid())
    
    
    if __name__ == '__main__':
        start = time.time()
        t = ThreadPoolExecutor(2)
        result = []
    
        for i in range(10):
            t_result = t.submit(work, i)
            result.append(t_result)
    
        for res in result:
            print(res.result())
    
        print(time.time() - start)

*  全局锁：

python 中的全局锁是 GIL 全局锁，GIL 是全局解释器锁，其它语言中没有这个 GIL。

其它语言中，同一个进程的线程可能会在不同的 cpu 上；而 python 中，同一个进程的线程只会在同一个 cpu 上。这样的差异就是 GIL 全局锁导致的，GIL 全局锁是 python 解释器自动加上的。

因为 GIL 全局锁的缘故，python 中进程的线程只能在单一 cpu 工作，这就限制了多线程的性能。

但 GIL 全局锁也保证了线程的安全，可以使用多进程 + 多线程的方式来弥补被限制的性能。

*  异步：

正常情况下，主程序是从上到下同步执行的。异步，相对同步而言，它不会影响或者不会阻塞主程序的执行。同步意味着有序，异步意味着无序，正因为异步的无序，使得各个程序间的协调成为一大难题。

异步与多进程多线程类似，但也有不同。异步与多进程多线程的区别：

异步是轻量级的线程，也称为协程
    
    可以获取异步函数的返回值，而多进程多线程无法获取返回值
    
    异步必须保证在主进程是异步的情况下才可以使用，而多进程多线程随时可以创建使用
    
    异步更适合文件读写使用，多进程多线程更适合业务的处理

在 python 中，关键字 `async` 用来定义一个异步，在一个异步程序里使用关键字 `await` 用来调用另一个异步。关键字 `await` 只能调用经过关键字 `async` 声明过的函数。

用法：

async def func1():
        return 返回值
        
    async def func2():
        result = await func1()

在主程序中，通过 `asyncio` 模块来调用 `async` 函数。`asyncio` 模块的常用方法：

<table> 
 <thead> 
  <tr> 
   <th>函数名</th> 
   <th>介绍</th> 
   <th>参数</th> 
   <th>返回值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>gather</td> 
   <td>将异步函数批量执行</td> 
   <td>async_func,…</td> 
   <td>List 或 函数的返回结果</td> 
  </tr> 
  <tr> 
   <td>run</td> 
   <td>执行主异步函数</td> 
   <td>[task]</td> 
   <td>执行函数的返回结果</td> 
  </tr> 
 </tbody> 
</table>

示例：

# coding:utf-8
    
    import os
    import time
    import random
    import asyncio
    
    
    async def a():
        for i in range(10):
            print(i, 'a', os.getpid())
            await asyncio.sleep(random.random() * 2)
    
        return 'a'
    
    
    async def b():
        for i in range(10):
            print(i, 'b', os.getpid())
            await asyncio.sleep(random.random() * 2)
    
        return 'b'
    
    
    async def main():
        result = await asyncio.gather(
            a(),
            b()
        )
    
        print(result)
    
    
    if __name__ == '__main__':
        start = time.time()
        asyncio.run(main())
        print('parent pid is %s' % os.getpid())
        print(time.time() - start)

`gevent` 模块也可以调用 `async` 函数。通过 `pip install gevent` 来安装 `gevent` 模块。`gevent` 模块的常用方法：

<table> 
 <thead> 
  <tr> 
   <th>函数名</th> 
   <th>介绍</th> 
   <th>参数</th> 
   <th>返回值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>spawn</td> 
   <td>创建协程对象</td> 
   <td>func,args</td> 
   <td>协程对象</td> 
  </tr> 
  <tr> 
   <td>joinall</td> 
   <td>批量处理协程对象</td> 
   <td>[spawn_obj]</td> 
   <td>[spawn_obj]</td> 
  </tr> 
 </tbody> 
</table>

示例：

# coding:utf-8
    
    import os
    import time
    import random
    import gevent
    
    
    def a():
        for i in range(10):
            print(i, 'a', os.getpid())
            gevent.sleep(random.random() * 2)
    
        return 'a'
    
    
    def b():
        for i in range(10):
            print(i, 'b', os.getpid())
            gevent.sleep(random.random() * 2)
    
        return 'b'
    
    
    if __name__ == '__main__':
        start = time.time()
        gevent_a = gevent.spawn(a)
        gevent_b = gevent.spawn(b)
        gevent_list = [gevent_a, gevent_b]
        result = gevent.joinall(gevent_list)
        print(result[0].value)
        print(result[1].value)
        print('parent pid is %s' % os.getpid())
        print(time.time() - start)

--------------------