既然 Celery 4 已经放弃了 Windows 支持,Windows 上的 (Python 3) 任务队列的最佳选择是什么?
What is the best option for a (Python 3) task queue on Windows now that Celery 4 has dropped Windows support?
我们 运行 Windows 上 IIS 下的一个 Flask 站点,对于进程外任务,我们使用 Celery。 Celery 在 Windows 下给我们带来了一些问题,但目前我们对 运行ning 版本 3.1.12 感到满意,使用 RabbitMQ/AMQP 作为后端,在 Windows 下工作.
新版本的 Celery (4) 有 dropped support for Windows,所以我正在寻找一个可行的替代方案。
RQ 貌似是一个很不错的任务队列,但是也不支持Windows(bottom of the page)
我还看到了一些看似不太受欢迎的任务队列,例如:
但不清楚这些是否支持 Windows 和 Flask。
我想知道是否有人有过 运行 在 Windows 下设置 Python 任务队列的经验。也许是我提到的其中之一,或者替代方案。
我们 运行 一台 Linux 机器不是一个选择,因为我们没有管理 Linux 的经验,而且我们有很多遗留的东西 运行需要Windows.
的宁
我 运行 Flask with Huey on Windows 没有任何问题,诚然仅用于开发和测试。对于生产,我在 Linux 服务器上使用 Flask/Huey。两者都有 Redis 后端,Flask 0.12 和 Huey 1.2.0 .
我使用工厂模式创建了专门的 "cut down" 版本的 Flask 应用程序,供 Huey 任务特定使用。此版本不加载蓝图或配置 Flask-Admin,因为 Huey 任务不需要这些。
应用程序文件夹中 __init__.py
的示例代码。 App
是从 Flask
:
延伸而来的 class
def create_app(settings_override=None):
app = App('app')
if settings_override:
app.config.from_object(settings_override)
else:
app.config.from_object(os.environ['APP_SETTINGS'])
from .ext import configure_extensions
configure_extensions(app, admin, load_modules=True)
# REST
import rest.api_v1
app.register_blueprint(api_v1_bp, url_prefix='/api/v1')
# ... and more suff
def create_huey_app():
app = App('huey app')
app.config.from_object(os.environ['APP_SETTINGS'])
from .ext import configure_extensions
configure_extensions(app, admin=None, load_modules=False)
return app
configure_extensions
的想法取自 Quokka CMS. Examine its app
__init__.py
and its extensions module,看看它是如何实现的。另请注意该项目如何创建特定应用程序 (create_celery_app
) 以用于 Celery 任务队列。
tasks.py
的示例。注意使用 with app.app_context():
来创建 Flask 上下文。现在我的函数可以访问 Flask-Mail、Flask-SqlAlchemy(数据库、模型)等扩展。
@huey.task()
def generate_transaction_documents_and_email(transaction_id):
app = create_huey_app()
with app.app_context():
reports.generate_transaction_documents_and_email(transaction_id)
@huey.task()
def send_email(subject, recipients, text_body, html_body, attachments=[], cc=[]):
app = create_huey_app()
with app.app_context():
emails.send_email(subject, recipients, text_body, html_body, attachments, cc)
@huey.periodic_task(crontab(minute='30'))
def synchronize_mailing_list():
app = create_huey_app()
if app.config['CREATESEND_SYNCHRONIZE']:
_list_name = app.config['CREATESEND_LIST']
with app.app_context():
sync_delete_ar_subscribers(_list_name)
sync_add_ar_subscribers(_list_name)
我们 运行 Windows 上 IIS 下的一个 Flask 站点,对于进程外任务,我们使用 Celery。 Celery 在 Windows 下给我们带来了一些问题,但目前我们对 运行ning 版本 3.1.12 感到满意,使用 RabbitMQ/AMQP 作为后端,在 Windows 下工作.
新版本的 Celery (4) 有 dropped support for Windows,所以我正在寻找一个可行的替代方案。
RQ 貌似是一个很不错的任务队列,但是也不支持Windows(bottom of the page)
我还看到了一些看似不太受欢迎的任务队列,例如:
但不清楚这些是否支持 Windows 和 Flask。 我想知道是否有人有过 运行 在 Windows 下设置 Python 任务队列的经验。也许是我提到的其中之一,或者替代方案。
我们 运行 一台 Linux 机器不是一个选择,因为我们没有管理 Linux 的经验,而且我们有很多遗留的东西 运行需要Windows.
的宁我 运行 Flask with Huey on Windows 没有任何问题,诚然仅用于开发和测试。对于生产,我在 Linux 服务器上使用 Flask/Huey。两者都有 Redis 后端,Flask 0.12 和 Huey 1.2.0 .
我使用工厂模式创建了专门的 "cut down" 版本的 Flask 应用程序,供 Huey 任务特定使用。此版本不加载蓝图或配置 Flask-Admin,因为 Huey 任务不需要这些。
应用程序文件夹中 __init__.py
的示例代码。 App
是从 Flask
:
def create_app(settings_override=None):
app = App('app')
if settings_override:
app.config.from_object(settings_override)
else:
app.config.from_object(os.environ['APP_SETTINGS'])
from .ext import configure_extensions
configure_extensions(app, admin, load_modules=True)
# REST
import rest.api_v1
app.register_blueprint(api_v1_bp, url_prefix='/api/v1')
# ... and more suff
def create_huey_app():
app = App('huey app')
app.config.from_object(os.environ['APP_SETTINGS'])
from .ext import configure_extensions
configure_extensions(app, admin=None, load_modules=False)
return app
configure_extensions
的想法取自 Quokka CMS. Examine its app
__init__.py
and its extensions module,看看它是如何实现的。另请注意该项目如何创建特定应用程序 (create_celery_app
) 以用于 Celery 任务队列。
tasks.py
的示例。注意使用 with app.app_context():
来创建 Flask 上下文。现在我的函数可以访问 Flask-Mail、Flask-SqlAlchemy(数据库、模型)等扩展。
@huey.task()
def generate_transaction_documents_and_email(transaction_id):
app = create_huey_app()
with app.app_context():
reports.generate_transaction_documents_and_email(transaction_id)
@huey.task()
def send_email(subject, recipients, text_body, html_body, attachments=[], cc=[]):
app = create_huey_app()
with app.app_context():
emails.send_email(subject, recipients, text_body, html_body, attachments, cc)
@huey.periodic_task(crontab(minute='30'))
def synchronize_mailing_list():
app = create_huey_app()
if app.config['CREATESEND_SYNCHRONIZE']:
_list_name = app.config['CREATESEND_LIST']
with app.app_context():
sync_delete_ar_subscribers(_list_name)
sync_add_ar_subscribers(_list_name)