redis - redis:使用joblib批量插入

Question

我有一份记录 list 。对于我的每条记录，我都需要进行一些繁重的计算，因为我要在Redis中创建反向索引。为了达到到达记录，需要在管道中执行多个redis命令(sadd为100 s + set为1 s)。

我想并行化此索引创建部分(使用joblib)，但未能这样做。第一个问题是我想将redis连接传递给每个作业，但是由于joblib想要对其进行序列化而无法正常工作，因此它不起作用。因此，只需提交主机/端口并让每个流程创建它自己的连接。

def heavy_calc_insert(value, host, port)

    r = redis.Redis(host=host, port=port)

    #... some calc

    pipe = r.pipeline()

    for bit in bits:
        key = "bit:" + str(bit)
        pipe.sadd(key, idx_value)

    pipe.set(idx_value, id)
    pipe.execute()

Parallel(n_jobs=4)(delayed(heavy_calc_insert)(value, host, port) for value in values)

但是，有了这段代码，我很快就得到了 ConnectionError:

ConnectionError: Connection closed by server.

我假设我碰到了太多连接的一种形式。

我该如何解决这个问题？

Answer 1

使用连接池。 https://github.com/andymccurdy/redis-py/blob/master/README.rst#connection-pools

从heavy_calc_insert中创建池，然后传递池而不是主机和端口。

pool = redis.ConnectionPool(host=host, port=port, db=0)
...
def heavy_calc_insert(value, pool)
    r = redis.Redis(connection_pool=pool)
    ...

Redis具有默认的10,000个连接限制，并且不会关闭未使用的连接。

连接池将使创建的连接处于受控状态，并且还可以提高性能。