Python进程间通信(IPC)详解：管道、队列、共享内存与信号量

"""
pipe.py 管道通信
注意
1. multiprocessing 中管道通信只能用于有亲缘关系进程中
2. 管道对象在父进程中创建,子进程通过父进程获取
"""
from multiprocessing import Process, Pipe

# 创建 管道
fd1, fd2 = Pipe()
def app1():
    print('1 启动应用1,请登录')
    print('2 请求app2授权')
    fd1.send('app1 请求登录')
    data = fd1.recv()
    print('5 ??',data)
    if data:
        print('6 登录成功:', data)


def app2():
    print('2.5')
    # 阻塞等待读取管道内容
    data = fd2.recv()
    print('3 app2', data)
    fd2.send(('Dave', '123'))
    print('4 app2,发送完毕')


if __name__ == '__main__':
    print(-5)
    p1 = Process(target=app1)
    print(-4)
    p2 = Process(target=app2)
    print(-3)
    p1.start()
    print(-2)
    p2.start()
    print(-1)
    p1.join()
    print(0)
    p2.join()
    print('运行结束')
"""
运行结果
(base) [root@VM-0-12-centos pipe]# python pipe.py
-5
-4
-3
-2
-1
1 启动应用1,请登录
2 请求app2授权
2.5
3 app2 app1 请求登录
4 app2,发送完毕
5 ?? ('Dave', '123')
6 登录成功: ('Dave', '123')
0
运行结束

"""

"""
quque_test.py 消息队列演示
注意: 消息队列符合先进先出原则
"""

# 创建消息队列
from multiprocessing import Queue, Process
from random import randint

# 以下模拟 双色球
# 创建消息队列
q = Queue(5)

def handle():
    print(5)
    for i in range(6):
        q.put(randint(1, 33))
    q.put(randint(1, 16))

def request():
    print(6)
    l = [q.get() for i in range(6)]
    l.sort()
    l.append(q.get())
    print(7,l)

if __name__ == '__main__':
    print(0)
    p1 = Process(target=handle)
    print(1)
    p2 = Process(target=request)
    print(2)
    p1.start()
    print(3)
    p2.start()
    print(4)
    p1.join()
    p2.join()
    print(8)
    print('程序结束')

from multiprocessing import Value,Array

obj = Value(ctype,data)
功能: 开辟共享内存
参数: ctype 表示共享内存空间类型 'i' 'f' 'c'
	data 共享内存空间初始数据
返回值: 共享内存对象

obj.value 对象属性的修改查看即对共享内存读写

obj = Array(ctype,data)
功能: 开辟共享内存
参数: ctype 表示共享内存空间类型 'i' 'f' 'c'
	data 整数表示开辟空间的大小,其数据表示开辟空间
返回值: 共享内存对象

Array共享内存读写:通过遍历obj可以得到每个值,直接通过索引可以修改

* 可以使用obj.value 直接打印共享内存中的字节串

"""
value.py 开辟单一共享内存空间
注意:共享内存只能有一个值
"""

from multiprocessing import Value,Array,Process


import time
import random
ctype = 'i'
data = 5000
# 创建共享内存
money = Value(ctype,data)

# 操作共享内存
def man():
    for i in range(30):
        time.sleep(0.1)
        money.value += random.randint(1,1000)
        print('man',money.value)

def girl():
    for i in range(30):
        time.sleep(0.1)
        money.value -= random.randint(1,800)
        print('girl', money.value)

if __name__ == '__main__':
    p1 = Process(target=man)
    p2 = Process(target=girl)
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    print('程序结束',money.value)

"""
array.py 开辟单一共享内存空间
注意:共享内存只能有一个值
"""

from multiprocessing import Array,Process


ctype = 'i'
data = [1,2,3,4]
# 创建共享内存
# shm = Array(ctype,data)

# 表示初始值 [0,0,0,0,0]
shm = Array(ctype,5)


def fun():
    for i in shm:
        print(i)
    shm[1] = 1000
if __name__ == '__main__':
    p1 = Process(target=fun)
    p1.start()
    p1.join()
    print('程序结束',shm[1])

from multiprocessing import Semaphore

sem = Semaphore(num)
功能: 创建信号量对象
参数: 信号量的初始值
返回值: 信号量对象

sem.acquire() 信号量减1 当信号量为0时阻塞
sem.release() 信号量加1
sem.get_value() 获取信号量数量

"""
信号量.py 信号量演示
思路: 信号量数量相当于资源,执行任务必须消耗资源
"""
import os
from multiprocessing import Semaphore
from multiprocessing import Process
from time import sleep

num = 3
sem = Semaphore(num)


# 任务函数
def handle():
    sem.acquire()  # sem -1
    print("%s 开始执行任务" % os.getpid())
    sleep(3)
    print("%s 执行任务完毕" % os.getpid())
    sem.release()  # sem +1


if __name__ == '__main__':
    ths = []
    for i in range(10):
        p = Process(target=handle)
        p.start()
        ths.append(p)

    for p in ths:
        p.join()
    print('程序结束')