Python 多线程高效读取 Elasticsearch 数据

下面是一个使用 Python 多线程读取 ES 数据的示例代码：

from elasticsearch import Elasticsearch
from threading import Thread, Lock

es = Elasticsearch()

def search_data(index, query, result, lock):
    res = es.search(index=index, body=query)
    with lock:
        result.extend(res['hits']['hits'])

def multi_thread_search(index, query, thread_num=4):
    result = []
    lock = Lock()
    threads = []
    for i in range(thread_num):
        t = Thread(target=search_data, args=(index, query, result, lock))
        threads.append(t)
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return result

该代码首先定义了一个 search_data 函数，该函数使用 ES 的 search 方法来查询数据，并将查询结果添加到 result 列表中。由于多个线程同时操作 result 列表，因此需要使用锁来保证线程安全。

然后定义了一个 multi_thread_search 函数，该函数使用多个线程来并行查询 ES 数据。该函数接受三个参数：index 表示要查询的索引名称，query 表示查询语句，thread_num 表示要使用的线程数，默认为 4。

该函数首先创建一个空的 result 列表和一个锁对象。然后创建指定数量的线程，并将它们添加到 threads 列表中。接着启动每个线程，并等待它们全部结束。最后返回查询结果。

使用该函数可以方便地在多线程环境下读取 ES 数据，提高数据读取效率。