dify ≥ 1.8.0 版本起正式支持 rocketmq,需配置 dify_async_mq_type=rocketmq 并部署 rocketmq 5.1.4+,确保 nameserver/broker 连通、主题与消费组正确创建,否则将回退至同步执行。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜

确认Dify版本与RocketMQ兼容性
必须使用 Dify ≥ 1.8.0 版本,该版本起正式支持自定义消息中间件插槽,且已移除对 Celery 的硬依赖,允许通过 `DIFY_ASYNC_MQ_TYPE=rocketmq` 显式启用 RocketMQ 驱动。
低于 1.8.0 的版本无法识别 RocketMQ 配置项,强行写入将被忽略并默认回退至同步执行模式,【工作流不会真正异步,但日志无报错】。
部署RocketMQ服务并验证连通性
在生产环境部署 RocketMQ 5.1.4 或更高版本(推荐 5.2.0),确保 nameserver 和 broker 均正常运行,且网络策略放行 9876(nameserver)和 10911(broker)端口。
执行命令验证连通性:./tools.sh org.apache.rocketmq.example.quickstart.Producer,若输出包含 SendResult [sendStatus=SEND_OK] 即表示基础链路通畅;若提示 No route info of this topic,需先用 ./mqadmin updateTopic 创建主题 dify_workflow_events。
配置Dify连接RocketMQ参数
编辑 .env 文件,添加以下四行:
DIFY_ASYNC_MQ_TYPE=rocketmq
ROCKETMQ_NAMESRV_ADDR=192.168.10.20:9876
ROCKETMQ_TOPIC=dify_workflow_events
ROCKETMQ_GROUP=dify-consumer-group-v1
注意:ROCKETMQ_GROUP 必须全局唯一,同一 group 内多个 Dify 实例会竞争消费,导致任务丢失;若部署多套 Dify 环境,每套必须配不同 group 名。
注册RocketMQ驱动并启用异步工作流
第一步:在 extensions/ext_async_mq/__init__.py 中注册驱动类
第二步:导入 rocketmq.client 并定义 RocketMQAsyncBroker 类,继承 BaseAsyncBroker,重写 publish() 和 subscribe() 方法
第三步:在 core/workflow/executor.py 中,将原 workflow.run() 调用替换为 broker.publish(workflow_id, inputs)
第四步:重启 Dify 服务,观察日志中是否出现 ✅ RocketMQ broker initialized with group: dify-consumer-group-v1
触发异步工作流并验证事件投递
调用 Dify API 启动工作流,设置 "response_mode": "async" 并携带 "callback_url":
curl -X POST "http://localhost:5001/v1/chat-messages" \
-H "Authorization: Bearer xxx" \
-d '{"inputs":{"query":"test"},"user":"u1","response_mode":"async","callback_url":"https://webhook.site/xxx"}'
成功后立即返回 JSON,含 "task_id" 字段;同时登录 RocketMQ 控制台或执行 ./mqadmin consumerProgress -g dify-consumer-group-v1,确认 MESSAGE_QUEUE_NUM 有增量且 CONSUMED_OFFSET 持续追平 MAX_OFFSET。











