Python请求Elastic API如何配置max_retries与retry_on_timeout

不需要强制使用官方Python Elasticsearch客户端，原生requests库本身就可以对接底层urllib3的重试能力，直接配置最大重试次数、超时重试规则，不需要额外依赖，具体实现如下：

具体实现代码 #

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 1. 配置重试规则
retry_config = Retry(
    total=3, # 即max_retries，最大重试总次数，可根据需求调整
    retry_on_timeout=True, # 开启超时自动重试
    allowed_methods=["POST"], # _bulk接口用的是POST，必须显式允许POST触发重试，默认配置只重试GET类幂等请求
    status_forcelist=[429, 500, 502, 503, 504], # 服务端返回这些状态码时也自动重试
    backoff_factor=1 # 重试退避系数，避免高频重试打崩ES，间隔计算规则为 系数 * 2^(已重试次数-1) 秒，不需要退避可以设为0
)

# 2. 初始化带重试能力的requests会话
session = requests.Session()
adapter = HTTPAdapter(max_retries=retry_config)
# 给http、https协议都挂上重试适配器
session.mount("http://", adapter)
session.mount("https://", adapter)

# 3. 替换原来的requests.post调用，用session发请求即可
headers = {'Accept': "application/json", 'Content-type': "application/json"}
# 注意必须显式传入timeout参数，否则请求会无限挂起，无法触发超时重试逻辑，这里示例设为30秒超时，可按需调整
respond = session.post(
    url=f"{ES_CLIENT}/{index_name}/_bulk",
    auth=auth,
    headers=headers,
    data=data_to_post,
    timeout=30
)

注意事项 #

• 给_bulk接口开POST重试前，一定要确认你的写入逻辑做了幂等处理，否则重试可能导致数据重复写入。
• 官方Elasticsearch客户端只是把这套重试逻辑做了封装，配置更简便，本质上和上面手动实现的逻辑没有差异，不是唯一选择。

内容的提问来源于stack exchange，提问作者user19435462