如何在使用多处理时将数据添加到 json 文件？

Question

我正在使用一个用户友好的基于 json 的面向文档的数据库，名为 TinyDB。但是我无法将多条数据添加到我的数据库中，因为我正在使用多处理。过了一会儿，我得到了数据库中已经存在 id x 的错误（这是因为 2 个或更多进程正在尝试同时添加数据）。有没有办法解决这个问题？

每次运行我都会插入新的唯一参数。

示例参数：

params = {'id' = 1, 'name': 'poop', 'age': 99}

代码：

resultsDb = TinyDB('db/resultsDb.json')

def run(params):
    resultsDb.insert({'id': params['id'], 'name': params['name'], 'age': params['age']})

maxProcesses = 12 # Cores in my pc

for i in range(maxProcesses):
    processes.append(Process(target=run, args=(params,)))

for p in processes:
    p.start()

for p in processes:
    p.join()

Answer 1

如果您想并行化数据写入，则需要将问题分解为更小的步骤，以便确保插入数据的步骤已经将所有内容合并在一起。这样你就不会在写入时遇到任何（明显的）线程安全问题。

例如，假设您的 JSON 文件具有三个字段，email、name 和 age，并且您希望对 email 强制唯一性，但有些记录是双输入的。例如，有一个条目有emily@smith.com 和她的名字，另一个有她的年龄。

您首先要制作一些东西来将所有内容组合在一起，然后并行化写入。

我会画一些代码（注意我没有测试过这个！）：

my_data = # some JSON data
grouped = {}

for datum in my_data:
    if datum['email'] in grouped:
        grouped[datum['email']].update(datum)
    else:
        grouped[datum['email']] = datum

# parallelize write as above

Answer 2

我无法在我有权访问的 Linux 系统上对此进行测试，因为它是一个共享服务器，运行代码所需的某些设施已被禁止访问。这是下面的 Windows 版本。但主要特点是：

1. 它使用Lock 来确保插入是序列化的，我认为这对于它运行而不会出错是必要的。当然，这违背了将代码并行化的目的，并且可以得出结论，使用多处理或多线程确实没有意义。
2. 在 Windows 中，我不必将 resultsDb = TinyDB('db.json') 语句移动到 run 函数内，因为在 spawn 用于创建新进程的平台上，例如 Windows，如果我将该语句留在全局范围内无论如何，它都会为每个新创建的进程执行。然而，对于 Linux，fork 用于创建新进程，它不会为每个新进程执行，而是每个新进程将继承由主进程打开的单个数据库。这可能有效，也可能无效——您可以在全局范围内尝试两种方式的语句。如果您将其放回全局范围以查看它是否在那里工作，您不需要在源代码的底部进行相同的声明。

from tinydb import TinyDB
from multiprocessing import Process, Lock


def run(lock, params):
    resultsDb = TinyDB('db/resultsDb.json')
    with lock:
        resultsDb.insert({'id': params['id'], 'name': params['name'], 'age': params['age']})
    print('Successfully inserted.')

# required by Windows:
if __name__ == '__main__':
    params = {'id': 1, 'name': 'poop', 'age': 99}

    maxProcesses = 12 # Cores in my pc

    lock = Lock()
    processes = []
    for i in range(maxProcesses):
        processes.append(Process(target=run, args=(lock, params)))

    for p in processes:
        p.start()

    for p in processes:
        p.join()

    # remove the following if the first one is at global scope:
    resultsDb = TinyDB('db/resultsDb.json')
    print(resultsDb.all())

打印：

Successfully inserted.
Successfully inserted.
Successfully inserted.
Successfully inserted.
Successfully inserted.
Successfully inserted.
Successfully inserted.
Successfully inserted.
Successfully inserted.
Successfully inserted.
Successfully inserted.
Successfully inserted.
[{'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}, {'id': 1, 'name': 'poop', 'age': 99}]