如何在Vercel部署集成Celery的Flask应用？RabbitMQ连接失败求助

问题背景 #

我正在开发一个集成Celery处理后台任务的Flask应用，本地环境下用命令celery -A index.celery worker -l info -P gevent启动RabbitMQ broker，运行完全正常。现在要把应用部署到Vercel，但不知道怎么操作。想问能不能用AWS上的第三方broker解决问题？如果没别的方案，我考虑用Vercel的ServerlessQ选项。

我试过在Vercel项目设置里把celery -A index.celery worker -l info -P gevent设为开发命令，但没生效。

Vercel报错信息：

[WARNING]   2023-09-17T13:12:25.821Z    52603464-4cb5-44dd-8e31-41304a5c6526    No hostname was supplied. Reverting to default 'localhost'
[ERROR] 2023-09-17T13:12:31.874Z    52603464-4cb5-44dd-8e31-41304a5c6526    Exception on / [POST]
Traceback (most recent call last):
  File "/var/task/kombu/utils/functional.py", line 32, in __call__
    return self.__value__
AttributeError: 'ChannelPromise' object has no attribute '__value__'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/var/task/kombu/connection.py", line 472, in _reraise_as_library_errors
    yield
  File "/var/task/kombu/connection.py", line 459, in _ensure_connection
    return retry_over_time(
  File "/var/task/kombu/utils/functional.py", line 318, in retry_over_time
    return fun(*args, **kwargs)
  File "/var/task/kombu/connection.py", line 934, in _connection_factory
    self._connection = self._establish_connection()
  File "/var/task/kombu/connection.py", line 860, in _establish_connection
    conn = self.transport.establish_connection()
  File "/var/task/kombu/transport/pyamqp.py", line 203, in establish_connection
    conn.connect()
  File "/var/task/amqp/connection.py", line 323, in connect
    self.transport.connect()
  File "/var/task/amqp/transport.py", line 129, in connect
    self._connect(self.host, self.port, self.connect_timeout)
  File "/var/task/amqp/transport.py", line 184, in _connect
    self.sock.connect(sa)
ConnectionRefusedError: [Errno 111] Connection refused

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/var/task/flask/app.py", line 2190, in wsgi_app
    response = self.full_dispatch_request()
  File "/var/task/flask/app.py", line 1486, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/var/task/flask/app.py", line 1484, in full_dispatch_request
    rv = self.dispatch_request()
  File "/var/task/flask/app.py", line 1469, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
  File "./index.py", line 18, in home
    compobador_ofertas.check_sale.delay(email, url)
  File "/var/task/celery/app/task.py", line 444, in delay
    return self.apply_async(args, kwargs)
  File "/var/task/celery/app/task.py", line 594, in apply_async
    return app.send_task(
  File "/var/task/celery/app/base.py", line 797, in send_task
    self.backend.on_task_call(P, task_id)
  File "/var/task/celery/backends/rpc.py", line 164, in on_task_call
    maybe_declare(self.binding(producer.channel), retry=True)
  File "/var/task/kombu/messaging.py", line 218, in _get_channel
    channel = self._channel = channel()
  File "/var/task/kombu/utils/functional.py", line 34, in __call__
    value = self.__value__ = self.__contract__()
  File "/var/task/kombu/messaging.py", line 234, in <lambda>
    channel = ChannelPromise(lambda: connection.default_channel)
  File "/var/task/kombu/connection.py", line 953, in default_channel
    self._ensure_connection(**conn_opts)
  File "/var/task/kombu/connection.py", line 459, in _ensure_connection
    return retry_over_time(
  File "/var/lang/lib/python3.9/contextlib.py", line 137, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/var/task/kombu/connection.py", line 476, in _reraise_as_library_errors
    raise ConnectionError(str(exc)) from exc
kombu.exceptions.OperationalError: [Errno 111] Connection refused

解决方案 #

为什么Vercel上直接启动Celery Worker无效 #

Vercel是Serverless平台，所有函数都是按需触发、执行完成即销毁的短期进程，而Celery Worker需要长期运行来监听任务队列，所以你在Vercel里设置的开发命令根本无法维持Worker进程，自然不会生效。另外报错里的Connection refused是因为Celery默认连接localhost的RabbitMQ，但Vercel环境里没有部署RabbitMQ服务。

方案一：使用AWS第三方Broker（完全可行） #

你可以用AWS上的托管Broker服务，搭配独立部署的Celery Worker来解决：

• 选择AWS Broker服务：比如Amazon MQ（原生支持RabbitMQ协议，和你本地用的兼容），或者Amazon ElastiCache for Redis（如果愿意切换Celery的Broker为Redis）。
• 配置连接信息：创建Broker实例后，获取它的连接URL，在Vercel项目的环境变量里设置CELERY_BROKER_URL和CELERY_RESULT_BACKEND（如果需要结果存储）。
• 调整Flask代码：修改Celery初始化逻辑，读取环境变量里的连接信息，不要硬编码localhost。
• 部署Celery Worker：因为Vercel不能运行长期进程，你需要把Worker部署到AWS的EC2、ECS容器服务，或者用Celery Lambda适配器把Worker包装成Lambda函数触发，也可以用托管的Worker服务比如Celery Cloud Tasks。

方案二：改用Vercel ServerlessQ #

如果不想额外管理Worker和Broker，ServerlessQ是更适配Vercel的方案，它是专为Vercel生态设计的异步任务服务：

• 注册并创建队列：在ServerlessQ平台注册账号，创建对应任务的队列。
• 替换Celery代码：安装ServerlessQ的Python SDK，把原来Celery的check_sale.delay(email, url)调用替换成ServerlessQ的任务提交API。
• 部署任务处理函数：把原来的Celery任务逻辑改成Vercel Serverless函数，ServerlessQ会自动触发这个函数处理队列里的任务。

内容的提问来源于stack exchange，提问作者Víctor M