多进程多线程修改全局变量

lizy169

如题
在各进程中修改（增加）全局变量，正在学习，应是各进程内存独立的问题，还没有搞明白
同时请教吾爱大神，看怎么做比较优雅点，谢谢大神指点！

[Python] 纯文本查看 复制代码

import time
import threading as td
import multiprocessing as mp
import pandas as pd
import numpy as np

df = pd.DataFrame()


def s_run():
    data = {"one": np.random.randn(4), "two": np.linspace(1, 4, 4), "three": ['zhangsan', '李四', 999, 0.1]}
    df1 = pd.DataFrame(data, index=[1, 2, 3, 4])
    global df
    df = pd.concat([df, df1], ignore_index=True)


def s_start():
    s1 = mp.Process(target=s_run)
    s2 = mp.Process(target=s_run)
    s3 = mp.Process(target=s_run)

    mp_list = [s1, s2, s3]
    for t in mp_list:
        t.start()
    for t in mp_list:
        t.join()

    while True:
        if not s1.is_alive():
            s1 = mp.Process(target=s_run)
            s1.start()
        if not s2.is_alive():
            s2 = mp.Process(target=s_run)
            s2.start()
        if not s3.is_alive():
            s3 = mp.Process(target=s_run)
            s3.start()
        for t in mp_list:
            t.join()

        time.sleep(1)


def p_run():
    print(len(df))


def p_start():
    p1 = mp.Process(target=p_run)
    p2 = mp.Process(target=p_run)
    p3 = mp.Process(target=p_run)
    mp_list = [p1, p2, p3]
    for t in mp_list:
        t.start()
    for t in mp_list:
        t.join()

    while True:
        if not p1.is_alive():
            p1 = mp.Process(target=p_run)
            p1.start()
        if not p2.is_alive():
            p2 = mp.Process(target=p_run)
            p2.start()
        if not p3.is_alive():
            p3 = mp.Process(target=p_run)
            p3.start()
        for t in mp_list:
            t.join()

        time.sleep(1)


def main():
    s = td.Thread(target=s_start)
    p = td.Thread(target=p_start)
    s.start()
    p.start()
    s.join()
    p.join()


if __name__ == '__main__':
    main()

bu^shan

[Python] 纯文本查看 复制代码

import time
import threading as td
import multiprocessing as mp
import pandas as pd
import numpy as np

class DataProcessor:
    def __init__(self, shared_dict, shared_counter):
        self.shared_dict = shared_dict
        self.shared_counter = shared_counter

    def s_run(self):
        data = {"one": np.random.randn(4), "two": np.linspace(1, 4, 4), "three": ['zhangsan', '李四', 999, 0.1]}
        df1 = pd.DataFrame(data, index=[1, 2, 3, 4])
        self.shared_dict.update(df1.to_dict())
        with self.shared_counter.get_lock():
            self.shared_counter.value += 1

    def s_start(self):
        processes = [mp.Process(target=self.s_run) for _ in range(3)]
        for p in processes:
            p.start()
        for p in processes:
            p.join()

    def p_run(self):
        print("Data length:", len(self.shared_dict))
        print("Update count:", self.shared_counter.value)

    def p_start(self):
        processes = [mp.Process(target=self.p_run) for _ in range(3)]
        for p in processes:
            p.start()
        for p in processes:
            p.join()

    def start_threads(self):
        s_thread = td.Thread(target=self.s_start)
        p_thread = td.Thread(target=self.p_start)
        s_thread.start()
        p_thread.start()
        s_thread.join()
        p_thread.join()

def main():
    manager = mp.Manager()
    shared_dict = manager.dict()
    shared_counter = manager.Value('i', 0) 

    processor = DataProcessor(shared_dict, shared_counter)
    processor.start_threads()

if __name__ == '__main__':
    main()

lizy169

主进程启动两个子线程；
一个子线程又启动三个进程负责数据收集，增加到全局变量中；
另一个子线程启动三个进程对收集的数据进行计算分析；
目前纯在的问题是进程中的数据增加不到全局变量中；

chenzhigang

你的全局变量是哪个

lizy169

chenzhigang 发表于 2024-7-15 16:34
你的全局变量是哪个

df，定义的一个空表格为全局变量

asd124689

封装成函数, 并加锁

马了顶大

去了解下进程间通信，有管道、socket等方式