python3--进程同步(multiprocess.Lock, Semaphore, Event)

import socket
from multiprocessing import Process
def chat(conn):
    while True:  # 接收消息，打印消息
        msg = conn.recv(1024).decode('utf-8')
        print(msg)
        conn.send((msg + '_sb').encode('utf-8'))
    conn.close()


if __name__ == '__main__':
    sk = socket.socket()
    sk.bind(('127.0.0.1', 9595))
    sk.listen()
    while True:  # 创建多进程
        conn, addr = sk.accept()
        Process(target=chat, args=[conn,]).start()
    sk.close()

import socket
sk = socket.socket()
sk.connect(('127.0.0.1', 9595))

while True:
    inp = input('>>>').encode('utf-8')
    sk.send(inp)
    msg = sk.recv(1024).decode('utf-8')
    print(msg)
sk.close()

import os
import time
import random
from multiprocessing import Process


def work(n):
    print('{} : {} is running'.format(n, os.getpid()))
    time.sleep(random.random())
    print('{} : {} is done'.format(n, os.getpid()))


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

import os
import time
import random
from multiprocessing import Process
from multiprocessing import Lock

def work(n, lock):
    lock.acquire()  #给进程上锁
    print('{} : {} is running'.format(n, os.getpid()))
    time.sleep(random.random())
    print('{} : {} is done'.format(n, os.getpid()))
    lock.release()  #给进程下锁


if __name__ == '__main__':
    lock = Lock()
    for i in range(3):
        p = Process(target=work, args=(i, lock))
        p.start()

import os
import time
import random
from multiprocessing import Process
def work(n):
    print('{} : {} is running'.format(n, os.getpid()))
    time.sleep(random.random())
    print('{} : {} is done'.format(n, os.getpid()))


if __name__ == '__main__':
    for i in range(3):
        p = Process(target=work, args=(i,))
        p.start()
        p.join()

# 文件db的内容为：{"count":3}
# 注意一定要用双引号，不然json无法识别
# 并发运行，效率高，但竞争写同一文件，数据写入错乱
import json
import time
import random
from multiprocessing import Process,Lock

def check_ticket(i):
    with open('ticket') as f:
        ticket_count = json.load(f)
    print('person{}查询当前余票 : '.format(i), ticket_count['count'])

def buy_ticket(i, lock):
    check_ticket(i)  # 调用查询余票函数
    lock.acquire()  # 锁进程
    with open('ticket') as f:
        ticket_count = json.load(f)
    time.sleep(random.random())  # 模拟读取延迟
    if ticket_count['count'] > 0:
        print('person{}购票成功'.format(i))
        ticket_count['count'] -= 1
        with open('ticket', 'w') as f:
            json.dump(ticket_count, f)
    else:
        print('余票不足，person{}购票失败'.format(i))
    time.sleep(random.random())  # 模拟写入延迟
    lock.release()  # 解锁进程

if __name__ == '__main__':
    lock = Lock()  # 实例化锁这个类得到一个对象lock
    for i in range(5):
        Process(target=buy_ticket, args=(i, lock,)).start()

#加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但
牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信，但问题是：
1.效率低（共享数据基于文件，而文件是硬盘上的数据）
2.需要自己加锁处理

#因此我们最好找寻一种解决方案能够兼顾：1、效率高（多个进程共享一块内存的数据）2、帮我们处理好锁问题。这就是
mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
队列和管道都是将数据存放于内存中
队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，
我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往
往可以获得更好的可获展性。

互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据 。
假设商场里有4个迷你唱吧，所以同时可以进去4个人，如果来了第五个人就要在外面等待，等到有人出来才能再进去玩。
实现：
信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，
acquire()调用被阻塞。这是迪科斯彻（Dijkstra）信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器
这样的有限资源。
信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念

import time
import random
from multiprocessing import Process, Semaphore

def ktv(i,sem):
    sem.acquire()  # 锁住进程
    print('person {} 进来唱歌了'.format(i))
    time.sleep(random.randint(1, 5))  # 模拟唱歌时间
    print('person {} 从ktv出去了'.format(i))
    sem.release()  # 解锁进程


if __name__ == '__main__':
    sem = Semaphore(4)  # 模拟只有4个房间
    for i in range(5):  # 模拟5批人需要唱歌
        Process(target=ktv, args=(i, sem)).start()

python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。
    事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，
    如果“Flag”值为True，那么event.wait 方法时便不再阻塞。
clear：将“Flag”设置为False
set：将“Flag”设置为True

from multiprocessing import Event

# 创建一个事件的对象
e = Event()
print(e.is_set())  # 在事件的创始之初，状态为False

from multiprocessing import Event

# 创建一个事件的对象
e = Event()
print(e.is_set())  # 在事件的创始之初，状态为False
e.wait()  # 阻塞，等待is_set()的值变成True，如果不为True，会一直等待
          # 后面的代码不会执行
print(1111)

from multiprocessing import Event

# 创建一个事件的对象
e = Event()
print(e.is_set())   # 打印is_set()的状态
                    # 在事件的创始之初，状态为False
e.set()  # 将is_set()的值改为True
e.wait()  # 阻塞，等待is_set()的值变成True，如果不为True，会一直等待
          # 后面的代码不会执行
print(e.is_set())  # 打印is_set()的状态

from multiprocessing import Event

# 创建一个事件的对象
e = Event()
print(e.is_set())   # 打印is_set()的状态
                    # 在事件的创始之初，状态为False
e.set()  # 将is_set()的值改为True
e.wait()  # 阻塞，等待is_set()的值变成True，如果不为True，会一直等待
          # 后面的代码不会执行
print(e.is_set())  # 打印is_set()的状态
e.clear()  # 将is_set()的值改为False
print(e.is_set())  # 打印is_set()的值
e.wait()  # 阻塞，等待is_set()的值变成True
print(111)  # 代码不会被执行，上面被阻塞了

import time
import random
from multiprocessing import Process, Event
def car(i, e):  # 感知状态的变化
    if not e.is_set():  # 当前这个事件的状态如果是False
        print('car{}正在等待'.format(i))  # 这辆车正在等待通过路口
    e.wait()  # 阻塞，直到有一个e.set行为(True)  # 理解为等红灯
    print('car{}通过路口'.format(i))

def traffic_light(e):  # 修改事件的状态
    print('33[1;31m红灯亮33[0m')
    # 事件在创始之初的状态是False,相当于程序中的红灯

    time.sleep(2)  # 红灯亮2秒
    while True:
        if not e.is_set():  # False
            print('33[1;32m绿灯亮33[0m')
            e.set()  # 修改is_set()的值为 True
        elif e.is_set():
            print('33[1;31m红灯亮33[0m')
            e.clear()  # 修改is_set()的值为 False
        time.sleep(2)  # 等待2秒


if __name__ == '__main__':
    e = Event()
    Process(target=traffic_light, args=(e,)).start()
    for i in range(10):
        time.sleep(random.randrange(0, 5, 2))
        Process(target=car, args=(i, e)).start()

Queue([maxsize]) 
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。
另外，还需要运行支持线程以便队列中的数据传输到底层管道中。 
Queue的实例q具有以下方法：

q.get( [ block [ ,timeout ] ] ) 
返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果
设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的
时间间隔内没有项目变为可用，将引发Queue.Empty异常。

q.get_nowait( ) 
同q.get(False)方法。

q.put(item [, block [,timeout ] ] ) 
将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，
将引发Queue.Empty异常（定义在Queue库模块中）。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发
Queue.Full异常。

q.qsize() 
返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加
或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。


q.empty() 
如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和
使用结果之间，队列中可能已经加入新的项目。

q.full() 
如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）。。

q.close() 
关闭队列，防止队列中加入更多数据。调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法
完成时马上关闭。如果q被垃圾收集，将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常
例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。

q.cancel_join_thread() 
不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。

q.join_thread() 
连接队列的后台线程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创
建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。

from multiprocessing import Queue, Process
# 能维护一个先进先出的次序，也能进行IPC

def haha(q):
    # 返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止
    print(q.get())


if __name__ == '__main__':
    q = Queue()
    p = Process(target=haha, args=(q, ))
    p.start()
    # 将[1, 2, 3]放入队列。如果队列已满，此方法将阻塞至有空间可用为止
    q.put([1, 2, 3])

import time
from multiprocessing import Queue, Process
# 能维护一个先进先出的次序，也能进行IPC

def haha(q):
    # 返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止
    print(q.get())
    # 将'你好'放入队列。如果队列已满，此方法将阻塞至有空间可用为止
    q.put('你好')


if __name__ == '__main__':
    q = Queue()
    p = Process(target=haha, args=(q, ))
    p.start()
    # 将'hi'放入队列。如果队列已满，此方法将阻塞至有空间可用为止
    q.put('hi')
    time.sleep(0.5)
    # 返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止
    print(q.get())

from multiprocessing import Process, Queue
import time, random, os


def consumer(q):
    while True:
        res = q.get()
        time.sleep(random.randint(1, 3))
        print('33[45m%s 吃 %s33[0m' % (os.getpid(), res))


def producer(q):
    for i in range(10):
        time.sleep(random.randint(1, 3))
        res = '包子%s' % i
        q.put(res)
        print('33[44m%s 生产了 %s33[0m' % (os.getpid(), res))


if __name__ == '__main__':
    q = Queue()
    # 生产者们:即厨师们
    p1 = Process(target=producer, args=(q,))

    # 消费者们:即吃货们
    c1 = Process(target=consumer, args=(q,))

    # 开始
    p1.start()
    c1.start()
    print('主')

from multiprocessing import Process, Queue
import time, random, os


def consumer(q):
    while True:
        res = q.get()
        if res is None: break  # 收到结束信号则结束
        time.sleep(random.randint(1, 3))
        print('33[1;32m%s 吃 %s33[0m' % (os.getpid(), res))


def producer(q):
    for i in range(10):
        time.sleep(random.randint(1, 3))
        res = '包子%s' % i
        q.put(res)
        print('33[1;31m%s 生产了 %s33[0m' % (os.getpid(), res))
    q.put(None)  # 发送结束信号


if __name__ == '__main__':
    q = Queue()
    # 生产者们:即厨师们
    p1 = Process(target=producer, args=(q,))

    # 消费者们:即吃货们
    c1 = Process(target=consumer, args=(q,))

    # 开始
    p1.start()
    c1.start()
    print('主')