Celery定时任务调用AWS S3 upload_file超时问题及解决诉求

问题描述 #

我编写了AWS S3文件上传方法，本地直接运行可正常上传文件，但整合到Celery定时任务执行时，调用s3_client.upload_file()出现连接超时。

上传方法代码 #

def upload_file_to_aws(file_name):
    """Upload a file to an S3 bucket
    :param file_name: File to upload
    :return: True if file was uploaded, else False
    """

    bucket = get_bucket()
    # If S3 object_name was not specified, use file_name
    object_name = file_name.split("/")[-1]
    
    # Upload the file
    s3_client = boto3.client(
        "s3",
        aws_access_key_id=AWS_ACCESS_KEY_ID,
        aws_secret_access_key=AWS_SECRET_ACCESS_KEY
    )
    ce_logger.info(f"s3_client:{s3_client}")
    try:
        s3_client.upload_file(file_name, bucket, object_name)
    except Exception as e:
        logger.info(e)
        return False
    return True

Celery日志报错 #

[the 2022-12-08 13:40:03, 205: INFO/ForkPoolWorker-1]  project.invoice.tasks.download_gmail_attachment_create_invoice[14b98680-7f38-45fb-9dbe-283330c304b0]: s3_client:<botocore.client.S3 object at 0x7f0a8a5f35e0>
[13:45:11 2022-12-08, 162: INFO/ForkPoolWorker-1]  project.invoice.tasks.download_gmail_attachment_create_invoice[14b98680-7f38-45fb-9dbe-283330c304b0]:  Connect timeout on endpoint URL: "https://karbon-text.s3.ap-south-1.amazonaws.com/upload_aSZcLIM.pdf"

Celery定时任务配置 #

app.conf.beat_schedule = {
    "download_gmail_attachment_create_invoice": {
        "task": "project.invoice.tasks.download_gmail_attachment_create_invoice",
        "schedule": crontab(minute="*/10"),
    },
}

任务函数代码 #

@shared_task
def download_gmail_attachment_create_invoice():
    file_name = get_file_name()
    upload_file_to_aws(file_name)

可能原因 #

• 网络环境限制：本地网络可正常访问S3，但Celery Worker所在服务器/容器的防火墙、安全组或VPC配置限制了出站流量到S3服务端口（443）。
• Fork进程连接池问题：Celery默认用Fork模式启动Worker，子进程会继承父进程的Socket连接，导致boto3客户端的连接池在子进程中无法正常工作。
• 未配置超时与重试：boto3默认超时参数较长，网络不稳定时易触发超时，且代码未设置重试机制。

解决方案 #

1. 验证网络连通性 #

登录Celery Worker所在服务器，执行命令测试S3 Endpoint连通性：

curl -v https://karbon-text.s3.ap-south-1.amazonaws.com

若无法连接，需：

• 调整服务器防火墙/安全组，允许出站访问443端口；
• 若在VPC环境，配置S3网关端点（Gateway Endpoint），让VPC内资源直接访问S3。

2. 优化boto3客户端初始化 #

在上传函数内延迟初始化客户端，并添加超时与重试配置，确保每个Worker进程使用独立的客户端实例：

s3_client = boto3.client(
    "s3",
    aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    config=boto3.session.Config(
        connect_timeout=10,  # 连接超时（秒）
        read_timeout=30,     # 读取超时（秒）
        retries={"max_attempts": 3}  # 重试次数
    )
)

3. 确保文件可访问性 #

确认get_file_name()返回的文件路径在Worker进程中存在且可读，避免临时文件被提前删除或Worker无读取权限。

4. 添加Celery任务重试机制 #

给任务绑定重试逻辑，遇到超时异常时自动重试：

@shared_task(bind=True, max_retries=3)
def download_gmail_attachment_create_invoice(self):
    file_name = get_file_name()
    try:
        upload_success = upload_file_to_aws(file_name)
        if not upload_success:
            raise Exception("S3上传失败")
    except Exception as e:
        self.retry(exc=e, countdown=60)  # 60秒后重试

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