如何在Django中配置Azure Service Bus作为Celery的Broker与结果后端

一、CELERY_BROKER_URL配置 #

Azure Service Bus支持队列、主题订阅两种模式，根据你的现有资源选择对应配置：

1. 使用现有队列作为Broker #

直接指定队列的完整URL，格式如下：

CELERY_BROKER_URL = "azureservicebus://<SAS_POLICY_NAME>:<SAS_KEY>@<SERVICE_BUS_NAMESPACE>.servicebus.windows.net/<QUEUE_NAME>"

替换占位符：

• <SAS_POLICY_NAME>：Azure Service Bus的共享访问策略名称（如RootManageSharedAccessKey或自定义策略）
• <SAS_KEY>：对应策略的主键/副键
• <SERVICE_BUS_NAMESPACE>：你的Service Bus命名空间（无需.servicebus.windows.net后缀）
• <QUEUE_NAME>：已创建的队列名称

2. 使用主题+订阅作为Broker #

Celery默认是点对点队列模式，若要使用主题订阅的发布订阅模式，需补充传输选项配置：

CELERY_BROKER_URL = "azureservicebus://<SAS_POLICY_NAME>:<SAS_KEY>@<SERVICE_BUS_NAMESPACE>.servicebus.windows.net"
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'topic_name': '<TOPIC_NAME>',
    'queue_name': '<SUBSCRIPTION_NAME>',
    'is_async': True,
}

替换占位符：

• <TOPIC_NAME>：已创建的主题名称
• <SUBSCRIPTION_NAME>：对应主题的订阅名称

二、CELERY_RESULT_BACKEND配置 #

Azure Service Bus是消息队列服务，不适合作为任务结果后端（缺乏持久化存储与结果检索的原生支持），推荐以下方案：

• 继续使用原有的Redis Cloud（无需改动，保持原有稳定配置）
• 迁移至Azure Redis Cache（适配Azure环境，配置格式与Redis Cloud一致）
• 使用Azure Blob Storage：需安装azure-storage-blob依赖，配置格式：

  CELERY_RESULT_BACKEND = "azureblockblob://<ACCOUNT_NAME>:<ACCOUNT_KEY>@<CONTAINER_NAME>"
  

• 使用Azure SQL/PostgreSQL等关系型数据库：配置为对应的数据库连接URL，例如：

  CELERY_RESULT_BACKEND = "postgresql://<USER>:<PASSWORD>@<HOST>:<PORT>/<DB_NAME>"
  

三、关键注意事项 #

• 权限验证：确保SAS策略拥有足够权限——队列模式需Send+Listen权限；主题订阅模式需对主题有Send权限、对订阅有Listen权限
• 版本兼容性：你的celery==5.2.7、kombu==5.2.4与azure-servicebus==7.11.0版本兼容，Kombu 5.2.x已原生支持Azure Service Bus v7+
• 网络检查：确认服务器IP已加入Azure Service Bus的防火墙允许列表，避免连接失败

四、验证步骤 #

1. 编写测试任务：

   from celery import Celery
   
   app = Celery('test_tasks', broker=CELERY_BROKER_URL, backend=CELERY_RESULT_BACKEND)
   
   @app.task
   def add(x, y):
       return x + y
   

2. 启动Celery Worker：

   celery -A test_tasks worker --loglevel=info
   

3. 调用测试任务并检查结果：

   result = add.delay(1, 2)
   print(result.get())  # 应输出3
   

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