Python进程间通信方式

from multiprocessing import Process, Manager
q = Manager().Queue(2)
q.put(1)
q.put(2,block=False,timeout=2)
def func():
    print(q.get())
 
p = Process(target=func)
print("size",q.qsize())
print("full",q.full())
p.start()
p.join()
print("empty",q.empty())
print("get", q.get())
print("get", q.get(block=False,timeout=2))

from multiprocessing import Process,Manager,active_children
import random
import queue
import time
 
class Producer(Process):
 
    def __init__(self,queue):
        super().__init__()
        self.queue = queue
 
    def run(self):
        for i in range(6):
            r = random.randint(0, 99)
            time.sleep(1)
            self.queue.put(r)
            print("add data{}".format(r))
 
class Consumer(Process):
 
    def __init__(self,queue):
        super().__init__()
        self.queue = queue
 
    def run(self):
        while True:
          if not self.queue.empty():
                data = self.queue.get()
                print("minus data{}".format(data))
 
 
if __name__ == '__main__':
    q = Manager().Queue() # 创建队列
    p = Producer(q)
    c = Consumer(q)
    p.start()
    c.start()
    print(active_children())  # 查看现有的进程
    p.join()
    c.join()
    print("结束")
 
 
>>>输出
[<ForkProcess(SyncManager-1, started)>, <Producer(Producer-2, started)>, <Consumer(Consumer-3, started)>]
add data83
minus data83
add data72
minus data72
add data8
minus data8
add data63
minus data63
add data75
minus data75
add data52
minus data52

from multiprocessing import Process,JoinableQueue
import os
import time
import random
 
 
def print_log(msg, log_type="prod"):
    if log_type == 'prod':
        print("\033[32;1m%s\033[0m" %msg)
    elif log_type == 'con':
        print("\033[31;1m%s\033[0m" %msg)
 
def producer(q):
    """
    生产者
    :param q: 
    :return: 
    """
    for i in range(10):
        data = random.randint(1,200)
        time.sleep(2)
        q.put(data)  # 放入队列
        msg = "add data {}".format(data)
        print_log(msg)
    q.join()  # 生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。
    # 阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止
 
 
 
 
def consumer(q):
    """
    消费者
    :param q: 
    :return: 
    """
    while True:
        if not q.empty():
            time.sleep(5)
            data = q.get()
            msg = "minus data{}".format(data)
            print_log(msg,"con")
            q.task_done()  # q.get()的返回项目已经被处理
 
 
if __name__ == '__main__':
    q = JoinableQueue()
    prod = Process(target=producer, args=(q,))
    con = Process(target=consumer, args=(q,))
    con.daemon = True  # 设置为守护进程，但是不用担心，producer内调用q.join保证了consumer已经处理完队列中的所有元素
    # 开启进程
    prod.start()
    con.start()
 
    prod.join()  # 等待生产和消费完成，主线程结束
    print("结束")