与多个Python脚本共享一个dict

我想dict从一个同时运行的多个Python脚本中访问一个唯一的(键/值)数据库.

如果script1.py更新d[2839],script2.py则应在查询几秒后查看修改后的值d[2839].

我想过使用SQLite,但似乎从多个进程并发写入/读取不是SQLite的强项(假设script1.py刚刚修改过d[2839],script2.pySQLite连接怎么会知道它必须重新加载数据库的这个特定部分？)

当我想要刷新修改时,我还考虑过锁定文件(但这样做相当棘手),并使用json.dump序列化,然后尝试检测修改,json.load如果有任何修改则用于重新加载等等......哦,不,我重新发明轮子,重新发明一个特别低效的键/值数据库!

redis看起来像一个解决方案,但它没有正式支持Windows,同样适用于leveldb.

多个脚本可能想要在完全相同的时间写入(即使这是一个非常罕见的事件),有没有办法让数据库系统处理这个(感谢一个锁定参数？似乎默认SQLite不能做这是因为"SQLite支持无限数量的同时读者,但它只会在任何时刻允许一位作者.")

什么是Pythonic解决方案呢？

注意:我在Windows上,并且dict应该有最多1M项(键和值都是整数).

1> georgexsh..：

---

除了SQLite之外的嵌入式数据存储的Mose没有并发访问的优化,我也对SQLite并发性能感到好奇,所以我做了一个基准测试:

import time
import sqlite3
import os
import random
import sys
import multiprocessing


class Store():

    def __init__(self, filename='kv.db'):
        self.cOnn= sqlite3.connect(filename, timeout=60)
        self.conn.execute('pragma journal_mode=wal')
        self.conn.execute('create table if not exists "kv" (key integer primary key, value integer) without rowid')
        self.conn.commit()

    def get(self, key):
        item = self.conn.execute('select value from "kv" where key=?', (key,))
        if item:
            return next(item)[0]

    def set(self, key, value):
        self.conn.execute('replace into "kv" (key, value) values (?,?)', (key, value))
        self.conn.commit()


def worker(n):
    d = [random.randint(0, 1<<31) for _ in range(n)]
    s = Store()
    for i in d:
        s.set(i, i)
    random.shuffle(d)
    for i in d:
        s.get(i)


def test(c):
    n = 5000
    start = time.time()
    ps = []
    for _ in range(c):
        p = multiprocessing.Process(target=worker, args=(n,))
        p.start()
        ps.append(p)
    while any(p.is_alive() for p in ps):
        time.sleep(0.01)
    cost = time.time() - start
    print(f'{c:<10d}\t{cost:<7.2f}\t{n/cost:<20.2f}\t{n*c/cost:<14.2f}')


def main():
    print(f'concurrency\ttime(s)\tpre process TPS(r/s)\ttotal TPS(r/s)')
    for c in range(1, 9):
        test(c)


if __name__ == '__main__':
    main()

结果在我的4核macOS盒,SSD卷上:

concurrency time(s) pre process TPS(r/s)    total TPS(r/s)
1           0.65    7638.43                 7638.43
2           1.30    3854.69                 7709.38
3           1.83    2729.32                 8187.97
4           2.43    2055.25                 8221.01
5           3.07    1629.35                 8146.74
6           3.87    1290.63                 7743.78
7           4.80    1041.73                 7292.13
8           5.37    931.27                  7450.15

结果在8核Windows Server 2012云服务器上,SSD容量:

concurrency     time(s) pre process TPS(r/s)    total TPS(r/s)
1               4.12    1212.14                 1212.14
2               7.87    634.93                  1269.87
3               14.06   355.56                  1066.69
4               15.84   315.59                  1262.35
5               20.19   247.68                  1238.41
6               24.52   203.96                  1223.73
7               29.94   167.02                  1169.12
8               34.98   142.92                  1143.39

事实证明,无论并发性如何,整体吞吐量都是一致的,并且SQLite在Windows上比macOS慢,希望这是有帮助的.

---

由于SQLite写锁是数据库方式,为了获得更多TPS,您可以将数据分区为多数据库文件:

class MultiDBStore():

    def __init__(self, buckets=5):
        self.buckets = buckets
        self.cOnns= []
        for n in range(buckets):
            cOnn= sqlite3.connect(f'kv_{n}.db', timeout=60)
            conn.execute('pragma journal_mode=wal')
            conn.execute('create table if not exists "kv" (key integer primary key, value integer) without rowid')
            conn.commit()
            self.conns.append(conn)

    def _get_conn(self, key):
        assert isinstance(key, int)
        return self.conns[key % self.buckets]

    def get(self, key):
        item = self._get_conn(key).execute('select value from "kv" where key=?', (key,))
        if item:
            return next(item)[0]

    def set(self, key, value):
        cOnn= self._get_conn(key)
        conn.execute('replace into "kv" (key, value) values (?,?)', (key, value))
        conn.commit()

在我的Mac上有20个分区的结果:

concurrency time(s) pre process TPS(r/s)    total TPS(r/s)
1           2.07    4837.17                 4837.17
2           2.51    3980.58                 7961.17
3           3.28    3047.68                 9143.03
4           4.02    2486.76                 9947.04
5           4.44    2249.94                 11249.71
6           4.76    2101.26                 12607.58
7           5.25    1903.69                 13325.82
8           5.71    1752.46                 14019.70

总TPS高于单个数据库文件.

---

@Basj - "在任何时刻只允许一个作者"是_cause,因为sqlite为你处理并发.如果一个进程正在写入,另一个尝试写入的进程不会失败,它将阻塞.您不需要为并发写入做任何事情就可以了.[本文档](https://sqlite.org/lockingv3.html#writing)描述了内部结构; [关于python库的这个](https://docs.python.org/2/library/sqlite3.html#sqlite3.connect)应该让你放心.