【操作系统知识】多线程、多进程、并发并行、锁的梳理

本是古典何须时尚 2023-02-24 11:23 26阅读 0赞

### 文章目录 ###

*   *  1. 并发与并行
     *   *  (1) 并发
         *  (2) 并行
         *  (3) 并发与并行的应用场景
     *  2. 同步与异步
     *  3. 阻塞与非阻塞
     *  10. 孤儿进程、僵尸进程、守护进程
     *  11. 进程池
     *  12. 守护线程
     *  13. 线程池
     *  14. 线程安全
     *  15. 多线程资源竞争，怎么解决
     *  16. 有几种锁

## 1. 并发与并行 ##

### (1) 并发 ###

> 并发是指同一时间间隔内多个任务都在运行，但是并不会在用一时刻同时运行，存在交替执行的情况，因此也经常存在资源竞争德尔情况。线程是并发的，实现的库有threading。

### (2) 并行 ###

> 并行指的是同一时刻多个任务同时运行，进程是并行的，实现的库有multiprocessing。

### (3) 并发与并行的应用场景 ###

> IO密集型操作（程序需要执行较多的读写，请求和回复任务的需要CPU对硬盘、内存进行读写，如爬虫）应使用并发更好，类似于多线程。

> CPU密集型操作（程序运行需要花大量时间做逻辑判断、运算等消耗CUP资源的操作，如复杂的浮点运算）应使用并行更好，类似于多进程。

## 2. 同步与异步 ##

> 同步和异步都是相对于多任务而言的。

> 同步：多个任务之间有先后顺序执行，一个执行完才能执行下一个。

> 异步：多个任务之间没有先后顺序，可以同时执行，有时候一个任务可能在必要的时候获取另一个同时执行的任务的结果，这叫回调。

## 3. 阻塞与非阻塞 ##

> 阻塞非阻塞是相对于代码执行而言的。

> 阻塞：如果卡住了调用者，调用者不能继续往下执行，当前线程会被挂起，调用线程只有在得到结果之后才会返回，就是说调用者阻塞了，函数只有在得到结果之后才会将阻塞的线程激活。

> 如果不会卡住，可以继续执行就是非阻塞的，非阻塞调用指在不能立刻得到结果之前也会立刻返回，同时该函数不会阻塞当前线程。

## 10. 孤儿进程、僵尸进程、守护进程 ##

> 孤儿进程：父进程退出，子进程还在运行这些子进程都是孤儿进程，孤儿进程将被init进程（进程号为1）说收养，并由init进程对他们完成状态收集工作。

> 僵尸进程：进程使用fork创建子进程，如果进程退出，而父进程并没有调用wait或waitpid获取子进程的状态信息，那么子进程的描述符仍然保存在系统中，这些进程就是僵尸进程。避免僵尸进程的方法：

*  fork两次用孙子进程去完成子进程的任务；
 *  用wait()函数使父进程阻塞；  
    \-使用信号量，在signal handler中调用waitpid，这样父进程不用阻塞。

> 守护进程：就是会随着主进程结束而结束的进程，主进程代码执行结束后就终止，而且守护进程内无法开启子进程，否则抛出异常。

import os
    import time
    from multiprocessing import Process
    
    class MyProcess(Process):
        def __init__(self, name):
            super().__init__()
            self.name = name
    
        def run(self):
            print(f"进程为{os.getpid()}，父进程为{os.getppid()}")
            print(f"我的名字是{self.name}")
    
    if __name__ == "__main__":
        p1 = MyProcess("张三")
        p2 = MyProcess("李四")
        p2.daemon = True  # 默认为False，必须在start()之前上设置
        p1.start()
        p2.start()
        time.sleep(2)
        print("主进程结束")
    
    # 结果
    '''
    进程为436，父进程为11972
    我的名字是张三
    进程为12140，父进程为11972
    我的名字是李四
    主进程结束
    '''

## 11. 进程池 ##

**[Pool进程池模块][Pool]**

> Multiprocessing.Pool可以提供指定数量的进程供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来执行它。Pool类用于需要执行的目标很多，而手动限制进程数量又太繁琐时，如果目标少且不用控制进程数量则可以用Process类。

> 进程池Pool语法：`Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])`

*  processes: 进程数，由CPU个数决定，不能操作CPU总数；如果进程是None，那么返回的数字`os.cpu_count()`。
 *  initializer: 如果是None，那么每一个工作进程在开始会调用`initialize(*initargs)`。
 *  maxtasksperchild: 工作进程退出前可以完成的任务数，完成后用一个新的工作进程来代替原进程，让闲置下来的资源被释放。默认值为None，表示只要Pool存在工作进程就回一直存活。
 *  context: 用在只当工作进程启动时的上下文，一般使用`multiprocessing.Pool()`或者一个`context()`对象的`Pool()`方法来创建一个池。

> multiprocessing 模块下的Pool类下的方法：

<table> 
 <thead> 
  <tr> 
   <th align="center">方法</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center"><code>apply(func[, args=()[, kwargs={}]])</code></td> 
   <td align="left">该函数用于传递不定参数，和python中的apply函数一致，主进程会被阻塞知道函数执行结束（不建议使用，python3已经移除）。</td> 
  </tr> 
  <tr> 
   <td align="center"><code>apply_async(func[, args=()[, kwargs={}[, callback[, error_callback]]]])</code></td> 
   <td align="left">与apply的用法一致，但是它是非阻塞的且支持结果返回后进行回调，调用失败，将调用<code>error_callback</code>。</td> 
  </tr> 
  <tr> 
   <td align="center"><code>map(func, iterable[, chunksize=None])</code></td> 
   <td align="left">Pool类中的map方法，与内置的map函数用法基本一致，它会使进程阻塞知道结果返回。（注意：虽然第二个参数是一个迭代器，但在实际使用中，必须在整个队列都就绪后，程序才会运行子进程。）</td> 
  </tr> 
  <tr> 
   <td align="center"><code>map_async(func, iterable[, chunksize[, callback[, error_callback]]])</code></td> 
   <td align="left">与map用法一致，但是它是非阻塞的。</td> 
  </tr> 
  <tr> 
   <td align="center"><code>imap(func, iterable[, chunksize])</code></td> 
   <td align="left">返回迭代器，<code>next()</code>调用返回的迭代器的方法得到结果，<code>imap()</code>方法有一个可选的超时参数：<code>next(timeout)</code>将引发<code>multiprocessing.TimeoutError</code>异常</td> 
  </tr> 
  <tr> 
   <td align="center"><code>close()</code></td> 
   <td align="left">防止任何更多的任务被提交到进程池中，一旦完成所有任务，工作进程将退出。</td> 
  </tr> 
  <tr> 
   <td align="center"><code>terminate()</code></td> 
   <td align="left">立即停止工作进程未完成的工作，当进程池对象被垃圾回收时，<code>terminate()</code>将立即调用。</td> 
  </tr> 
  <tr> 
   <td align="center"><code>join()</code></td> 
   <td align="left">等待工作进程退出，必须在<code>close()</code>或<code>terminate()</code>使用之前<code>join()</code>。</td> 
  </tr> 
 </tbody> 
</table>

from multiprocessing import Pool
    import time
    
    def run(fn):  # fn函数参数是数据列表的一个元素
        time.sleep(1)
        print(fn * fn)
    
    if __name__ == "__main__":
        s = [1, 2, 3, 4, 5, 6]
        print("执行顺序：")
        start = time.time()
        for fn in s:
            run(fn)
        end = time.time()
        print(f"顺序执行时间：{end-start:.2f}")
        print("创建多个进程：")
        pool = Pool(10)  # 创建拥有10个进程数量的进程池
        pool.map(run, s)
        pool.close()  # 关闭进程池，不在接收新的进程
        pool.join()  # 主进程阻塞等待子进程退出
        t = time.time()
        print(f"并行执行时间：{t-end:.2f}")
    
    # 结果
    '''
    执行顺序：
    1
    4
    9
    16
    25
    36
    顺序执行时间：6.00
    创建多个进程：
    1
    9
    16
    25
    4
    36
    并行执行时间：3.31
    '''

## 12. 守护线程 ##

> 守护线程：当主线程结束后，子线程也会随之结束，所以当主线程结束后，整个程序就退出了。所谓“线程守护”，就是主线程不管该线程的执行情况，只要是其他子线程结束且主线程执行完毕，主线程都会关闭。也就是说：主线程不等待该守护线程的执行完再去关闭。

> 守护线程的作用：是为其他线程提供便利服务，守护线程最典型的应用就是GC。

> 守护线程的特点：

*  只要当前主线程中尚存在任何一个非守护线程没有结束，守护线程就继续工作；
 *  只有当最后一个非守护线程结束时，守护线程才随着主线程一同结束工作。

#  线程不是守护线程
    import time
    import threading
    
    
    def fun():
        print("start fun")
        time.sleep(2)
        print("end fun")
    
    def main():
        print("main thread")
        t1 = threading.Thread(target=fun,args=())
        t1.setDaemon(False)  # 非守护线程
        t1.start()
        time.sleep(1)
        print("main thread end")
    
    if __name__ == '__main__':
        main()
    
    # 结果，主程序在等待子线程结束才退出
    '''
    main thread
    start fun
    main thread end
    end fun
    '''

# 线程是守护线程
    import time
    import threading
    
    
    def fun():
        print("start fun")
        time.sleep(2)
        print("end fun")
    
    def main():
        print("main thread")
        t1 = threading.Thread(target=fun,args=())
        t1.setDaemon(True)  # 守护线程
        t1.start()
        time.sleep(1)
        print("main thread end")
    
    if __name__ == '__main__':
        main()
    
    # 结果，程序在主程序结束后，直接退出。没有等待子线程运行完
    '''
    main thread
    start fun
    main thread end
    '''

# 多线程情况下
    import time
    import threading
    
    
    def fun():
        print("start fun")
        time.sleep(2)
        print("end fun")
    
    def foo():
        print("start foo")
        time.sleep(1)
        print("end foo")
    
    if __name__ == '__main__':
        print("main thread")
        t1 = threading.Thread(target=fun,args=())
        t2 = threading.Thread(target=foo,args=())
        t1.setDaemon(True)  # 守护线程
        t1.start()
        t2.start()
        time.sleep(1)
        print("main thread end")
    
    # 结果，程序在主程序结束后，等待非守护线程的foo执行完，才退出。而没有等待守护线程func运行完
    '''
    main thread
    start fun
    start foo
    main thread end
    end foo
    '''

## 13. 线程池 ##

**[PYTHON线程池及其原理和使用][PYTHON]**

> 系统启动一个新线程的成本是比较高的，因为它涉及与操作系统的交互。这种情形下，使用线程池可以很好地性能，尤其是当程序中需要大量存活期很短暂的线程时，更应该考虑使用线程池。

> 线程池在系统启动时创建大量空闲的线程，程序只要将一个函数提交给线程池，线程池就会启动一个空闲的线程来执行它。当该函数执行结束后，该线程并不会死亡，而是再次返回到线程池中变成空闲状态，等待执行下一个函数。

> 此外，使用线程池可以有效控制系统中并发线程的数量。当西戎中包含有大量的并发线程时，会导致系统性能急剧下降，甚至导致Python解释器崩溃，而线程池的最大线程数参数可以控制系统中并发线程的数量不超过此数。

**线程池的使用**

> 线程池的基类是`concurrent.futures`模块中的`Executor`，包含两个子类：`ThreadPoolExecutor`（用于创建线程池），`ProcessPoolExecutor`用于创建进程池。

> `Executor`中的常用方法：

<table> 
 <thead> 
  <tr> 
   <th align="center">方法</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center"><code>submit(func, *args, **kwargs)</code></td> 
   <td align="left">将函数func提交给线程池，<code>*args</code>表示传给func函数的参数，<code>**kwargs</code>表示以关键字形式给func传参</td> 
  </tr> 
  <tr> 
   <td align="center"><code>map(func, *iterables, timeout=None, chunksize=1)</code></td> 
   <td align="left">该函数类似于python中的map()函数的用法，只是个该函数将会启动多个线程，以异步的方式立即对可迭代对象iterables进行map处理</td> 
  </tr> 
  <tr> 
   <td align="center"><code>shutdown(wait=True)</code></td> 
   <td align="left">关闭线程池，不再接受性任务，会将以前所有提交的任务执行完成</td> 
  </tr> 
 </tbody> 
</table>

> 程序task函数提交(submit)给线程池后，submit方法会返回一个Future对象，Future类主要用于获取线程任务函数的返回值。由于线程任务会在新线程中以异步的方式执行，因此，线程执行的函数相当于一个“将来完成”的任务，所以Python使用Future来代表。

> `Future`中的常用方法：

<table> 
 <thead> 
  <tr> 
   <th align="center">方法</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center"><code>cancel()</code></td> 
   <td align="left">取消该Future代表的线程任务，若该任务正在执行，不可取消，返回False；否则，程序会取消该任务，并返回True</td> 
  </tr> 
  <tr> 
   <td align="center"><code>cancelled()</code></td> 
   <td align="left">返回Future代表的线程任务是否被成功取消</td> 
  </tr> 
  <tr> 
   <td align="center"><code>running()</code></td> 
   <td align="left">若该Future代表的线程任务正在执行，不可取消，该方法返回True</td> 
  </tr> 
  <tr> 
   <td align="center"><code>done()</code></td> 
   <td align="left">若该Future代表的线程任务被成功取消或者执行完成，该方法返回True</td> 
  </tr> 
  <tr> 
   <td align="center"><code>result(timeout=None)</code></td> 
   <td align="left">获取该Future代表的线程任务最后的返回的结果，若任务还未完成，该方法将会阻塞当前线程，其中timeout参数指定阻塞持续时间</td> 
  </tr> 
  <tr> 
   <td align="center"><code>exception(timeout=None)</code></td> 
   <td align="left">获取该Future代表的线程任务说引发的异常，若任务成功完成，且没有异常，该方法返回None</td> 
  </tr> 
  <tr> 
   <td align="center"><code>add_done_callback(func)</code></td> 
   <td align="left">为该Future代表的线程任务注册一个"回调函数"，当该任务成功完成时，程序会自动触发func函数 ,并将对应的 Future 对象作为参数传给该回调函数</td> 
  </tr> 
 </tbody> 
</table>

**使用线程池来执行线程任务的步骤：**

*  调用`ThreadPoolExecutor`类的构造器创建一个线程池；
 *  定义一个普通函数作为线程任务；
 *  调用`ThreadPoolExecutor`对象的`submit()`方法来提交线程任务；
 *  当不想提交任何任务时，调用`ThreadPoolExecutor`对象的`shutdown()`方法来关闭线程池

from concurrent.futures import ThreadPoolExecutor
    import threading
    import time
    
    # 定义一个线程任务函数
    def func(x, y):
        print(f"{threading.current_thread().name} is running, x={x}, y={y}.")
        # time.sleep(2)
        return x * y
    
    def get_result(future):
        """
        如果程序不希望直接调用 result() 方法阻塞线程，则可通过 Future 的 add_done_callback() 方法来添加回调函数，该回调函数形如 fn(future)。当线程任务完成后，程序会自动触发该回调函数，并将对应的 Future 对象作为参数传给该回调函数
        """
        print(future.result())
    
    def process():
        # 创建一个包含2个子线程的线程池，且线程名为以“thread_”开头
        pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread_")
        # 向线程池提交一个task
        for i in range(5):
            future = pool.submit(func, i, i+1)
            print(future.done())
            future.add_done_callback(get_result)
        pool.shutdown(wait=True)
        print('process finished')
    
    if __name__ == "__main__":
        process()
    
    # 结果
    '''
    thread__0 is running, x=0, y=1.
    False
    0
    thread__0 is running, x=1, y=2.
    True
    2
    False
    thread__0 is running, x=2, y=3.
    6
    False
    thread__1 is running, x=3, y=4.
    12
    False
    thread__0 is running, x=4, y=5.
    20
    process finished
    '''

"""
    使用 with 语句来管理线程池，这样即可避免手动关闭线程池，通过Exectuor里的map()方法，为 iterables 的每个元素启动一个线程，以并发方式来执行 action 函数，并收集线程任务的返回值
    """
    from concurrent.futures import ThreadPoolExecutor
    import threading
    import time
    
    # 定义一个准备作为线程任务的函数
    def action(max):
        my_sum = 0
        for i in range(max):
            print(threading.current_thread().name + '  ' + str(i))
            my_sum += i
        return my_sum
    # 创建一个包含4条线程的线程池
    with ThreadPoolExecutor(max_workers=4) as pool:
        # 使用线程执行map计算
        # 后面元组有3个元素，因此程序启动3条线程来执行action函数
        results = pool.map(action, (3, 5, 7))
        print('--------------')
        for r in results:
            print(r)
    
    # 结果
    '''
    ThreadPoolExecutor-0_0  0
    ThreadPoolExecutor-0_0  1
    ThreadPoolExecutor-0_0  2
    ThreadPoolExecutor-0_0  0
    ThreadPoolExecutor-0_0  1
    ThreadPoolExecutor-0_0  2
    ThreadPoolExecutor-0_0  3
    ThreadPoolExecutor-0_0  4
    ThreadPoolExecutor-0_0  0
    ThreadPoolExecutor-0_0  1
    ThreadPoolExecutor-0_0  2
    ThreadPoolExecutor-0_0  3
    ThreadPoolExecutor-0_0  4
    ThreadPoolExecutor-0_0  5
    ThreadPoolExecutor-0_0  6
    --------------
    
    3
    10
    21
    '''

**线程池的优点：**

*  降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
 *  提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
 *  提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

## 14. 线程安全 ##

> 线程安全就是多线程访问时，采用了加锁机制，当一个线程访问该类的某个数据时，进行保护，其他线程不能进行访问知道该线程读完，其他线程才能使用，这样就不会出现数据不一致或数据污染。

> 当多个线程访问某个方法时，不管通过怎样的调用方式或说这些线程如何交替执行，在主程序中不需要去做任何的同步，这个类的结果行为都是我们设想的正切行为，那么我们就可以说这个类是线程安全的。

## 15. 多线程资源竞争，怎么解决 ##

> 线程是独立的，同一个进程里线程的数据共享的，当各个线程访问数据资源时会出现竞争状态，即数据几乎同时被多个线程占用，造成数据混乱，所谓的线程的不安全。

> 添加锁，确保了某段关键代码（共享数据资源）只能由一个线程从头到尾完整地执行，则能够解决多线程资源进程下原子操作的问题；但是添加锁阻止了多线程并发操作，包含锁的某段代码实际上只能以单线程模式执行，效率就大大下降；其次，由于可以存在多个锁，不同的线程持有不同的说，并试图获取对方持有的锁时，可能造成死锁，导致多个线程全部挂起，既不能执行，也无法结束，只能靠操作系统强制终止。

## 16. 有几种锁 ##

*  互斥锁

> 在编程中，引入了对象互斥锁的概念，来保证共享数据操作的完整性。每个对象都对应于一个可称为" 互斥锁" 的标记，这个标记用来保证在任一时刻，只能有一个线程访问该对象。同一个进行中的多线程是共享系统资源的,，多个线程同时对一个对象进行操作，一个线程操作尚未结束，另一个线程已经对其进行操作，导致最终结果出现错误，此时需要对被操作对象添加互斥锁，保证每个线程对该对象的操作都得到正确的结果。

*  可重入锁

> 可重入锁，也叫做递归锁，指的是在同一线程内，外层函数获得锁之后，内层递归函数仍然可以获取到该锁。换一种说法，同一个线程再次进入同步代码时，可以使用自己已获取到的锁，防止在同一线程中多次获取锁而导致死锁发生。

*  死锁

> 若干个线程资源竞争时，都在等待对方对某部分资源解除占用状态，结果谁也不愿意先解锁，互相干等着，程序无法执行下去，这就是死锁。

**持续跟新，更多操作系统基础知识尚需补充，[详见个人博客归纳][Link 1]**。

**欢迎留言补充，批评指正！**

[Pool]: https://blog.csdn.net/brucewong0516/article/details/85788202
[PYTHON]: https://www.cnblogs.com/hoojjack/p/10846010.html
[Link 1]: https://holysll.github.io/2020/05/08/%E6%93%8D%E4%BD%9C%E7%B3%BB%E7%BB%9F%E7%9F%A5%E8%AF%86/