并发编程——进程——进程的同步与数据共享

进程的同步

一、互斥锁

进程之间的数据是不共享的，但是啊，你想，我们的代码运行在同一台电脑上，所以共享同一套文件系统，试想一下，如果一个进程想删除文件，另一个进程同时想读取文件。

举个简单的例子，我们让几个进程竞争标准输出终端：

import os
from multiprocessing import Process


def work():
    for item in range(2):
        print('%s %s is running.' % (os.getpid(), item))


if __name__ == '__main__':
    for i in range(3):
        process = Process(target=work)
        process.start()

我们希望的输出当然是：

21020 0 is running.
21020 1 is running.
19912 0 is running.
19912 1 is running.
5932 0 is running.
5932 1 is running.

但也有可能标准输出并不按照我们的预期：

8212 0 is running.
8212 1 is running.
18976 0 is running.18960 0 is running.

18960 1 is running.18976 1 is running.

这是因为啥，PID为8212的进程先执行，并且执行完毕，但PID为18976的进程第一次打印还没有结束的时候，PID为18960的进程开始了第一次输出。

一个简单的场景就是抢票，某一时间可能有很多很多人买同一张票，那么这张票到底该卖给谁，这是一个问题，现在来利用互斥锁模拟一下抢票：

票文件为tickets.json:

{"departure": "China", "destination": "American", "tickets": 5}

import time
import json
import random
from multiprocessing import Lock
from multiprocessing import Process


def search(name):
    tickets = json.load(open("ticket.json"))
    time.sleep(random.randint(0, 3))
    print(name, "查询到——》", "出发地：", tickets["departure"], "目的地：", tickets["destination"], "剩余票数：", tickets["tickets"])


def get(name):
    tickets = json.load(open("ticket.json"))
    time.sleep(random.randint(1, 3))
    if tickets["tickets"] > 0:
        tickets["tickets"] -= 1
        time.sleep(random.randint(1, 2))
        json.dump(tickets, open("ticket.json", "w"))
        print(name, "购票成功！")


def task(name, lock):
    search(name)
    with lock:
        get(name)


if __name__ == '__main__':
    locker = Lock()
    for i in range(7):
        person = "Alex %s" % i
        process = Process(target=task, args=(person, locker))
        process.start()

模拟结果为：

Alex 1 查询到——》 出发地： China 目的地： American 剩余票数： 5
Alex 4 查询到——》 出发地： China 目的地： American 剩余票数： 5
Alex 3 查询到——》 出发地： China 目的地： American 剩余票数： 5
Alex 6 查询到——》 出发地： China 目的地： American 剩余票数： 5
Alex 0 查询到——》 出发地： China 目的地： American 剩余票数： 5
Alex 2 查询到——》 出发地： China 目的地： American 剩余票数： 5
Alex 1 购票成功！
Alex 5 查询到——》 出发地： China 目的地： American 剩余票数： 5
Alex 4 购票成功！
Alex 3 购票成功！
Alex 6 购票成功！
Alex 0 购票成功！

枷锁可以保证多个进程想要修改同一块数据时只能有一个进程可以进行修改，即串行地修改，虽然速度变慢了，但是确保了数据安全。

二、生产者消费者模型介绍

为什么要使用生产者消费者模型

生产者指的是生产数据的任务，消费者指的是处理数据的任务。

在并发编程中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。

同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。

为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者和消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。

生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

这个阻塞队列就是用来给生产者和消费者解耦的。

三、生产者消费者模型实现

import time
import random
from multiprocessing import Process
from multiprocessing import JoinableQueue


def consumer(q, name):
    while True:
        time.sleep(random.randint(1, 3))
        print(name, "eat", q.get())
        q.task_done()   # 发送信号给q.join()说明已经从队列中取走一个数据并处理完毕了


def producer(q, name, food):
    for i in range(3):
        time.sleep(random.randint(2, 5))
        result = "%s %s" % (food, i)
        q.put(result)
        print(name, "creat", result)
    q.join()    # 等到消费者把自己放入队列中的所有的数据都取走之后，生产者才结束


if __name__ == '__main__':
    queue = JoinableQueue()

    producer1 = Process(target=producer, args=(queue, "producer 1", "红烧肉"))
    producer1.start()
    producer2 = Process(target=producer, args=(queue, "producer 2", "黄焖鸡"))
    producer2.start()
    producer3 = Process(target=producer, args=(queue, "producer 3", "铁板烧"))
    producer3.start()

    consumer1 = Process(target=consumer, args=(queue, "Alex"))
    consumer1.daemon = True
    consumer1.start()
    consumer2 = Process(target=consumer, args=(queue, "Coco"))
    consumer2.daemon = True
    consumer2.start()

    producer1.join()
    producer2.join()
    producer3.join()

输出结果为：

producer 3 creat 铁板烧 0
Coco eat 铁板烧 0
producer 1 creat 红烧肉 0
Alex eat 红烧肉 0
producer 2 creat 黄焖鸡 0
producer 3 creat 铁板烧 1
Coco eat 黄焖鸡 0
producer 1 creat 红烧肉 1
producer 2 creat 黄焖鸡 1
Alex eat 铁板烧 1
producer 3 creat 铁板烧 2
Coco eat 黄焖鸡 1
Alex eat 红烧肉 1
Coco eat 铁板烧 2
producer 1 creat 红烧肉 2
Coco eat 红烧肉 2
producer 2 creat 黄焖鸡 2
Alex eat 黄焖鸡 2

进程间数据共享

在同一台计算机上可以用文件共享数据实现进程间数据通信，但问题是，读写文件效率太低了，而且还需要自己加锁，加着加着估计自己都懵逼了，到底哪个锁开了，哪个锁关了。

这时候我们就需要一种能够在进程间通信机制：队列和管道。

队列

队列就像一个特殊的列表，但是可以设置固定长度，并且从前面插入数据，从后面取出数据，先进先出。

导入方式：

from multiprocessing import Queue

我们来看一下Queue的源码

class Queue(object):

    def __init__(self, maxsize=0, *, ctx):
        if maxsize <= 0:
            # Can raise ImportError (see issues #3770 and #23400)
            from .synchronize import SEM_VALUE_MAX as maxsize
        self._maxsize = maxsize
        self._reader, self._writer = connection.Pipe(duplex=False)
        self._rlock = ctx.Lock()
        self._opid = os.getpid()
        if sys.platform == 'win32':
            self._wlock = None
        else:
            self._wlock = ctx.Lock()
        self._sem = ctx.BoundedSemaphore(maxsize)
        # For use by concurrent.futures
        self._ignore_epipe = False

        self._after_fork()

        if sys.platform != 'win32':
            register_after_fork(self, Queue._after_fork)

可以看到--init--方法中有一个参数maxsize，代表初始化队列的时候可以传入最大存储数量。

    def put(self, obj, block=True, timeout=None):
        assert not self._closed, "Queue {0!r} has been closed".format(self)
        if not self._sem.acquire(block, timeout):
            raise Full

        with self._notempty:
            if self._thread is None:
                self._start_thread()
            self._buffer.append(obj)
            self._notempty.notify()

put方法，如果队列没有关闭并且没有满，追加一个传入的obj对象。

如果队列已满，此方法将阻塞至队列有空间可用为止。

block控制阻塞行为，默认为True，如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。

timeout指定在阻塞模式中等待可用空间的时间长短，超时后将引发Queue.Full异常。

    def get(self, block=True, timeout=None):
        if block and timeout is None:
            with self._rlock:
                res = self._recv_bytes()
            self._sem.release()
        else:
            if block:
                deadline = time.monotonic() + timeout
            if not self._rlock.acquire(block, timeout):
                raise Empty
            try:
                if block:
                    timeout = deadline - time.monotonic()
                    if not self._poll(timeout):
                        raise Empty
                elif not self._poll():
                    raise Empty
                res = self._recv_bytes()
                self._sem.release()
            finally:
                self._rlock.release()
        # unserialize the data after having released the lock
        return _ForkingPickler.loads(res)

get方法可以从队列首获取一个对象。

如果队列为空，此方法将阻塞，直到队列中有对象可用为止。

block用于控制阻塞行为，默认为True，如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。

timeout是可选超时时间，用在阻塞模式中，如果在制定的时间间隔内没有对象变为可用，将引发Queue.Empty异常。

    def get_nowait(self):
        return self.get(False)

    def put_nowait(self, obj):
        return self.put(obj, False)

get_nowait：就相当于get(False)，非阻塞。

put_nowait：就相当于put(obj, False)，非阻塞。

    def qsize(self):
        # Raises NotImplementedError on Mac OSX because of broken sem_getvalue()
        return self._maxsize - self._sem._semlock._get_value()

返回队列中目前项目的正确数量。

此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。

在某些系统上，此方法可能引发NotImplementedError异常。

    def empty(self):
        return not self._poll()

    def full(self):
        return self._sem._semlock._is_zero()

empty方法：如果调用此方法时 q为空，返回True。

如果其他进程或线程正在往队列中添加项目，结果是不可靠的。

也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。

full：如果q已满，返回为True。

由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）。

    def close(self):
        self._closed = True
        try:
            self._reader.close()
        finally:
            close = self._close
            if close:
                self._close = None
                close()

关闭队列，防止队列中加入更多数据。

调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。

如果队列被垃圾收集，将自动调用此方法。

关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。

例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。

来上手敲一敲，试试队列啥样：

from multiprocessing import Queue

if __name__ == '__main__':
    queue = Queue(7)
    for i in range(7):
        queue.put(i)
    try:
        queue.put_nowait(7)
    except:
        print("队列已满！")
    print("队列是否已满：", queue.full())
    print("queue get:", queue.get())
    print("queue get:", queue.get())
    print("queue get:", queue.get())
    print("queue get:", queue.get())
    print("队列是否已空：", queue.empty())
    try:
        queue.put_nowait(7)
    except:
        print("队列已满！")
    print("queue size =", queue.qsize())
    try:
        print("queue get nowait:", queue.get_nowait())
    except:
        print("队列已空！")
    print("队列是否已满：", queue.full())
    print("queue get:", queue.get())
    print("queue get:", queue.get())
    print("queue get:", queue.get())
    print("队列是否已空：", queue.empty())
    try:
        queue.get_nowait()
    except:
        print("队列已空！")

输出结果为：

队列已满！
队列是否已满： True
queue get: 0
queue get: 1
queue get: 2
queue get: 3
队列是否已空： False
queue size = 4
queue get nowait: 4
队列是否已满： False
queue get: 5
queue get: 6
queue get: 7
队列是否已空： True
队列已空！

在进程的一些操作中队列是安全的：同一时间只能一个进程拿到队列中的一个数据，你拿到了一个数据，这个数据别人就拿不到了。