并发编程

引入

• 面试题: 进程和线程的区别和联系
• 进程 - 操作系统分配内存的的基本单位 - 一个进程可以包含一个或多个线程
• 线程 - 操作系统分配CPU的基本单位

并发编程(concurrent programming)

1. 改善性能 - 让程序中没有因果关系的部分可以并发的执行
2. 改善用户体验 - 让耗时间的操作不会造成程序的假死

Python中实现并发的三种手段

1. 多线程 - 多个线程可以共享操作系统分配给进程的内存 - 线程通信相对简单
2. 多进程 - 进程之间的内存是不共享的 - 如果要通信需要使用IPC机制 - 管道/套接字/中间件
3. 异步I/O - 非阻塞式的I/O操作, 通过回调函数(钩子函数)对操作结果进行处理

• Redis/Node.js - 多路I/O复用(I/O操作是多路并发的)

import threading
from PIL import Image
import glob
import os

PREFIX = 'thumbnail'


def generate_thumbnail(infile, size, format='jpg'):
    size = 128, 128
    file, ext = os.path.splitext(infile)
    file = file[file.rfind('/') + 1:]
    outfile = f'{PREFIX}/{file} {size[0]} {size[1]}.{ext}'
    img = Image.open(infile)
    img.thumbnail(size, Image.ANTIALIAS)
    img.save(outfile, format)


def main():
    if not os.path.exists(PREFIX):
        os.mkdir(PREFIX)
    for infile in glob.glob('./thumbnail/*.jpg'):
        for size in (32, 64, 128):
            threading.Thread(
                target=generate_thumbnail, args=(infile, (size, size))
            ).start()
    # generate_thumbnail('D:\\PSL\\images\\u=3151669337,1400973154&fm=26&gp=0.jpg')


if __name__ == '__main__':
    main()

多个线程竞争一个资源时会发生什么

• 多线程程序如果在没有竞争资源的场景那么通常会比较简单
• 临界资源 - 被多个线程竞争的资源
• 当多个线程竞争临界资源的时候如果缺乏必要的保护措施就会导致混乱

import threading
import time
from concurrent.futures import ThreadPoolExecutor


class Account(object):
    '''银行帐户'''

    def __init__(self, lock):
        self.balance = 0.0
        self.lock = lock

    def deposit(self, money):

        # self.lock.acquire()
        # try:
        with self.lock:
            new_balance = self.balance + money
            time.sleep(0.01)
            self.balance = new_balance
        # finally:
        #     self.lock.release()


# 自定义线程类
class AddMoneyThread(threading.Thread):

    def __init__(self, account, money):
        super(AddMoneyThread, self).__init__()
        self.account = account
        self.money = money

    # run方法时线程启动之后要执行的回调方法(钩子函数)
    # 所以启动线程不能够直接调用run方法而是通过start方法启动线程
    # 什么时候需要使用回调式编程?
    # 你知道要做什么但不知道什么时候会做这件事请
    def run(self):
        # 通过锁保护临界资源
        # 可以写try-finally也可以使用上下文语法
        # self.lock.acquire()
        # try:
        with self.lock:
            new_balance = self.acount.balance + self.money
            time.sleep(0.01)
            self.acount.balance = new_balance
        # finally:
        #     self.lock.release()


def add_money(account, money):
    account.deposit(money)


def main():
    # lock = threading.Lock()
    # account = Account(lock)
    # threads = []
    # for _ in range(100):
    #     t = threading.Thread(target=add_money, args=(account, 1))
    #     threads.append(t)
    #     t.start()
    # for t in threads:
    #     t.join()
    # print(account.balance)

    #线程池
    lock = threading.Lock()
    account = Account(lock)
    pool = ThreadPoolExecutor(max_workers=100)
    futures = []
    for _ in range(100):
        future = pool.submit(add_money, account, 1)
        futures.append(future)
    for future in futures:
        future.result()
    print(account.balance)
    # 关闭线程池(此方法不阻塞, 会等到线程池中的线程都执行完后才会关闭线程池)
    pool.shutdown()


if __name__ == '__main__':
    main()

守护线程/守护进程 - daemon

• mysqld/httpd/firewalld/systemd
• 守护进程/守护线程是不值得为之保留程序的进程或线程
• 如果主线程结束了守护线程也不再保留即使守护线程还在执行(没有结束)
• 自定义线程类中要设置守护线程: super().init(demon=True)
• Lock - 多个线程竞争临界资源(资源只有一个) - 获得锁才能操作资源
• Condition - 基于Lock对昂可以创建Condition - wait()/notify_all() - 实现线程调度
• Semaphore - 多个线程竞争资源(资源有多个, 但线程数多于资源数, 拿走减1, 释放加1)
• 多个线程通信比较简单, 因为可以共享内存
• 多个线程通信相对比较困难, 可以使用multiprocessing.Queue, 通过多个继承共享一个队列来实现进程间的通信
• 生产者消费者模型

import threading
import time


def show_message(content):
    while True:
        print(content, end='')


def main():
    threading.Thread(target=show_message, args=('ping',), daemon=True).start()
    threading.Thread(target=show_message, args=('pong',), daemon=True).start()
    time.sleep(5)


if __name__ == '__main__':
    main()