Python python3进程间通信示例

文章目录 [隐藏]

python进程间共享数据的方式有：socket、文件/数据库、内存。内存方式即python中封装的特定功能的类。
python的进程也有同线程一样的Lock，避免操作同一对象时发生脏读。
这里演示Manager、Queue、Pipe三种方式。

1.Manager共享

1】自定义一个进程myprocess.py ：

import multiprocessing
import time


class MyProcess(multiprocessing.Process):
    def __init__(self, name, listor, mngList):
        super().__init__()
        self.name = name
        self.listor = listor
        self.mngList = mngList

    def run(self):
        self.show()

    def show(self):
        while True:
            print("进程1 -%s, 默认列表：%s，\tMng列表：%s，%s" % (self.pid, len(self.listor), len(self.mngList), self.name))
            time.sleep(2)

import multiprocessing

import time

class MyProcess(multiprocessing.Process):

def __init__(self, name, listor, mngList):

super().__init__()

self.name = name

self.listor = listor

self.mngList = mngList

def run(self):

self.show()

def show(self):

while True:

print("进程1 -%s, 默认列表：%s，\tMng列表：%s，%s" % (self.pid, len(self.listor), len(self.mngList), self.name))

time.sleep(2)

2】程序代码：

import multiprocessing
import os
import time
from myprocess import MyProcess

# 进程间默认不共享全局变量
globalList = []
# 使用Manager定义 可在进程间共享的全局变量
managerList = multiprocessing.Manager().list()


# 写操作进程
def process2():
    while True:
        print("进程2 -%s，默认列表：%s，\tMng列表：%s" % (os.getpid(), len(globalList), len(managerList))) # globalList递增1
        globalList.append(1)  # 操作全局变量
        managerList.append(1) # 操作全局变量
        time.sleep(0.5)


# 读操作进程
def process3():
    while True:
        print("进程3 -%s，默认列表：%s，\tMng列表：%s" % (os.getpid(), len(globalList), len(managerList)))  # globalList不变，一直是0
        time.sleep(0.5)


# 主进程
def main():
    MyProcess("MyProcess", globalList, managerList).start()
    p1 = multiprocessing.Process(target=process2) # 定义一个进程
    p2 = multiprocessing.Process(target=process3)
    p1.start() # 启动进程
    p2.start()


if "__main__" == __name__:
    # 默认主进程
    print("主进程启动")

    main() # 创建子进程

    while True:
        print("主程0 -%s，默认列表：%s，\tMng列表：%s" % (os.getpid(), len(globalList), len(managerList)))
        time.sleep(0.5)

import multiprocessing

import os

import time

from myprocess import MyProcess

# 进程间默认不共享全局变量

globalList = []

# 使用Manager定义可在进程间共享的全局变量

managerList = multiprocessing.Manager().list()

# 写操作进程

def process2():

while True:

print("进程2 -%s，默认列表：%s，\tMng列表：%s" % (os.getpid(), len(globalList), len(managerList))) # globalList递增1

globalList.append(1) # 操作全局变量

managerList.append(1) # 操作全局变量

time.sleep(0.5)

# 读操作进程

def process3():

while True:

print("进程3 -%s，默认列表：%s，\tMng列表：%s" % (os.getpid(), len(globalList), len(managerList))) # globalList不变，一直是0

time.sleep(0.5)

# 主进程

def main():

MyProcess("MyProcess", globalList, managerList).start()

p1 = multiprocessing.Process(target=process2) # 定义一个进程

p2 = multiprocessing.Process(target=process3)

p1.start() # 启动进程

p2.start()

if "__main__" == __name__:

# 默认主进程

print("主进程启动")

main() # 创建子进程

while True:

print("主程0 -%s，默认列表：%s，\tMng列表：%s" % (os.getpid(), len(globalList), len(managerList)))

time.sleep(0.5)

2.Queue共享

import multiprocessing
import time

# 队列。先进先出
queuer = multiprocessing.Queue(maxsize=5)  # 元素数量上限。最大长度5

# queue.put(item, block=True, timeout=None)方法参数：
#   item，放入队列中的数据元素。
#   block，当队列中元素个数达到上限(queuer.full()返回true)，继续往里放数据时：
#       如果 block=False，直接引发queue.Full异常；效果同put_nowait(item)方法。
#       如果 block=True-默认，timeout=None，则一直等待,直到出现空位放入数据；
#       如果 block=True，timeout=N(正整数)，则等待N秒。N秒后没出现空位则引发queue.Full异常。
#   timeout，设置超时时间。默认None。

# queue.get(block=True, timeout=None)方法参数：
#   block，当队列中没有数据元素(queuer.empty()返回true)，继续取数据时：
#       如果 block=False，直接引发queue.Empty异常；
#       如果 block=True-默认，timeout=None，则一直等待出现数据可以取出；
#       如果 block=True，timeout=N(正整数)，则等待N秒，N秒后没有数据可供取出则引发queue.Empty异常。
#   timeout，设置超时时间。默认None


def processPut():
    queuer.put(666)
    while True:
        queuer.put("A") 
        time.sleep(2) #  延迟2秒
        # print("添加：%s" %  queuer.qsize()) # Raises NotImplementedError on Mac OSX because of broken sem_getvalue()
        print("添加了")


def processGet():
    element = queuer.get()
    print(element)
    while True:
        ele = queuer.get() # 如果已经空了，则阻塞等待出现可用数据
        time.sleep(0.2)
        print("取出了：%s" % ele) 


def main():
    putor = multiprocessing.Process(target=processPut)
    getor = multiprocessing.Process(target=processGet)
    putor.start()
    getor.start()


if "__main__" == __name__:
    main()

import multiprocessing

import time

# 队列。先进先出

queuer = multiprocessing.Queue(maxsize=5) # 元素数量上限。最大长度5

# queue.put(item, block=True, timeout=None)方法参数：

# item，放入队列中的数据元素。

# block，当队列中元素个数达到上限(queuer.full()返回true)，继续往里放数据时：

# 如果 block=False，直接引发queue.Full异常；效果同put_nowait(item)方法。

# 如果 block=True-默认，timeout=None，则一直等待,直到出现空位放入数据；

# 如果 block=True，timeout=N(正整数)，则等待N秒。N秒后没出现空位则引发queue.Full异常。

# timeout，设置超时时间。默认None。

# queue.get(block=True, timeout=None)方法参数：

# block，当队列中没有数据元素(queuer.empty()返回true)，继续取数据时：

# 如果 block=False，直接引发queue.Empty异常；

# 如果 block=True-默认，timeout=None，则一直等待出现数据可以取出；

# 如果 block=True，timeout=N(正整数)，则等待N秒，N秒后没有数据可供取出则引发queue.Empty异常。

# timeout，设置超时时间。默认None

def processPut():

queuer.put(666)

while True:

queuer.put("A")

time.sleep(2) # 延迟2秒

# print("添加：%s" % queuer.qsize()) # Raises NotImplementedError on Mac OSX because of broken sem_getvalue()

print("添加了")

def processGet():

element = queuer.get()

print(element)

while True:

ele = queuer.get() # 如果已经空了，则阻塞等待出现可用数据

time.sleep(0.2)

print("取出了：%s" % ele)

def main():

putor = multiprocessing.Process(target=processPut)

getor = multiprocessing.Process(target=processGet)

putor.start()

getor.start()

if "__main__" == __name__:

main()

3.Pipe共享

import multiprocessing
import time

# 管道。进出随意但有方向
piper = multiprocessing.Pipe(True)  # True：双工-同时收发。Fase：左收右发，两个口是元组 [0]接收recv，[1]发送send
 
def processPut(): 
    for i in ["a", "b", "c", "d"]:
        piper[1].send(i) 
        print("添加了 ", i)
        time.sleep(2)


def processGet(): 
    while True:
        ele = piper[0].recv() # 如果已经空了，则阻塞等待出现可用数据 
        print("取出了：", ele) 
        # time.sleep(3)


def main():
    putor = multiprocessing.Process(target=processPut)
    getor = multiprocessing.Process(target=processGet)
    putor.start()
    getor.start()

    putor.join()       # 执行send时阻塞管道，使recv方法等待执行
    getor.terminate() # 循环send完毕，管道释放。recv取不到数据时，自动结束


if "__main__" == __name__:
    main()

import multiprocessing

import time

# 管道。进出随意但有方向

piper = multiprocessing.Pipe(True) # True：双工-同时收发。Fase：左收右发，两个口是元组 [0]接收recv，[1]发送send

def processPut():

for i in ["a", "b", "c", "d"]:

piper[1].send(i)

print("添加了 ", i)

time.sleep(2)

def processGet():

while True:

ele = piper[0].recv() # 如果已经空了，则阻塞等待出现可用数据

print("取出了：", ele)

# time.sleep(3)

def main():

putor = multiprocessing.Process(target=processPut)

getor = multiprocessing.Process(target=processGet)

putor.start()

getor.start()

putor.join() # 执行send时阻塞管道，使recv方法等待执行

getor.terminate() # 循环send完毕，管道释放。recv取不到数据时，自动结束

if "__main__" == __name__:

main()

-end

声明

本文由 cuiweiyou 原创，转载请注明出处：http://www.gaohaiyan.com/2661.html

承接App定制、企业web站点、办公系统软件设计开发，外包项目，毕设

Android系统广播和接收、自定义广播、无序广播
使用 aapt 查看apk文件版本信息
Java中的MD5加密
Android模拟器发布* failed to start daemon *ADB server didn’t ACK