22

我们在 Windows 上的 IIS 下运行 Flask 站点，对于进程外任务，我们使用 Celery。Celery 在 Windows 下给我们带来了一些问题，但现在我们对运行 3.1.12 版本感到满意，使用 RabbitMQ/AMQP 作为后端，它可以在 Windows 下运行。

新版本的 Celery (4) 已经放弃了对 Windows 的支持，所以我正在寻找一个可行的替代方案。

RQ 似乎是一个非常不错的任务队列，但它也不支持 Windows（页面底部）

我看到了更多，似乎不太受欢迎的任务队列，例如：

但目前尚不清楚这些是否支持 Windows 和 Flask。我想知道是否有人有在 Windows 下运行 Python 任务队列的经验。也许是我提到的其中一个，或者是替代方案。

运行 Linux 机器不是我们的选择，因为我们没有管理 Linux 的经验，而且我们有很多需要 Windows 运行的遗留系统。

4

1 回答 1

15

我在 Windows 上使用 Huey 运行 Flask 没有任何问题，诚然仅用于开发和测试。对于生产，我在 Linux 服务器上使用 Flask/Huey。两者都使用 Redis 后端，Flask 0.12 和 Huey 1.2.0。

我使用工厂模式来创建一个专门的“精简”版本的 Flask 应用程序，供 Huey 任务使用。此版本不加载蓝图或配置 Flask-Admin，因为 Huey 任务中不需要这些。

__init__.pyapp 文件夹中的示例代码。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。检查其app __init__.py及其扩展模块以了解其实现方式。还要注意这个项目如何创建一个特定的应用程序 ( create_celery_app) 用于 Celery 任务队列。

的例子tasks.py。注意with app.app_context():创建 Flask 上下文的使用。现在我的函数可以访问诸如 Flask-Mail、Flask-SqlAlchemy (db, models) 等扩展。

@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)
于 2017-03-25T18:19:39.310 回答