azure - 工作人员使用 celery、redis 和 rabbitMQ 发回结果的 2 分钟延迟
问题描述
我有什么:
我用 rabbitMQ 作为代理运行 celery,redis 作为结果后端。我有一个应用程序发送任务和处理任务的工作人员。
我将其部署如下:
- 该应用程序、redis、rabbitMQ 和一个工作人员(我们称他为“local_worker”)使用 docker-compose 在 azure VM 上运行,因此我使用 rabbitMQ 和 redis(6.2.5)的 docker 版本。rabbitMQ 和 redis 端口在 VM 上打开,并且这些容器配置了用户名和密码。
- 我使用连接到在 VM 上运行的 redis 和 rabbitMQ 的 azure 容器实例添加工作人员。
首先,如果您对此架构有建议,我很乐意得到建议。
问题:
一切正常,任务被分派给不同的工人,这些工人发回结果等等……
当任务在 30 分钟后发送而没有任务运行时,我观察到当任务未发送到“local_worker”时,redis 延迟为 2 分钟。
- 我知道这一定来自redis,因为我可以在发送任务后立即在worker容器实例中看到任务的日志。
- 我用flower和graphana和celery prometheus exporter监控这个架构,这样我就可以监控任务的延迟。在开花时,潜在任务保持“处理”状态。
- 在无任务间隔后的第一个任务上还有 120 秒的时间,并且未由“local_worker”处理。
- 当任务由与 redis 在同一 VM 上运行的“local_worker”处理时,不会发生这种情况。
这就像 redis 或虚拟机在发回结果之前休眠了 2 分钟。因为它正好是 120 秒(2 分钟),所以我希望它是 redis、celery 或 azure 想要的东西(确定性的东西)
我不使用 redis conf 文件,仅使用默认设置(密码除外)来运行 redis 服务器。
感谢您对我的架构和问题的帮助和反馈。
第一个和第三个任务已经由本地工人处理。第二个已由外部工作人员处理。在外部工作人员的日志上,我在返回结果之前放了一条打印线,这条线已在 14:14:23 打印。所以从这次打印到任务正式结束,已经有 120 秒的时间了。
编辑:
我发现 redis_socket_timeout 的默认值是 120 秒。
我删除了该行redis_retry_on_timeout = True
并redis_socket_keepalive = True
在我的 celery 配置文件中添加了该行。现在我得到的错误是任务失败了redis.exceptions.TimeoutError: Timeout reading from socket
。我不知道为什么套接字超时而结果已经准备好。我的容器实例的网络有问题吗?
这是我的码头工人撰写:
version: "3.5"
services:
rabbitmq:
image: rabbitmq:3.8-management
restart: always
ports:
- 5672:5672
labels:
- traefik.enable=true
- traefik.http.services.rabbitmq-ui.loadbalancer.server.port=15672
- traefik.http.routers.rabbitmq-ui-http.entrypoints=http
- traefik.http.routers.rabbitmq-ui-http.rule=(Host(`rabbitmq.${HOSTNAME?Variable not set}.sowit.app`))
- traefik.docker.network=traefik-public
- traefik.http.routers.rabbitmq-ui-https.entrypoints=https
- traefik.http.routers.rabbitmq-ui-https.rule=Host(`rabbitmq.${HOSTNAME?Variable not set}.sowit.app`)
- traefik.http.routers.rabbitmq-ui-https.tls=true
- traefik.http.routers.rabbitmq-ui-https.tls.certresolver=le
- traefik.http.routers.rabbitmq-ui-http.middlewares=https-redirect
env_file:
- .env
environment:
- RABBITMQ_DEFAULT_USER=${RABBITMQ_DEFAULT_USER}
- RABBITMQ_DEFAULT_PASS=${RABBITMQ_DEFAULT_PASS}
networks:
- traefik-public
redis:
image: redis:6.2.5
restart: always
command: ["redis-server", "--requirepass", "${RABBITMQ_DEFAULT_PASS:-password}"]
ports:
- 6379:6379
networks:
- traefik-public
flower:
image: mher/flower:0.9.5
restart: always
labels:
- traefik.enable=true
- traefik.http.services.flower-ui.loadbalancer.server.port=5555
- traefik.http.routers.flower-ui-http.entrypoints=http
- traefik.http.routers.flower-ui-http.rule=Host(`flower.${HOSTNAME?Variable not set}.sowit.app`)
- traefik.docker.network=traefik-public
- traefik.http.routers.flower-ui-https.entrypoints=https
- traefik.http.routers.flower-ui-https.rule=Host(`flower.${HOSTNAME?Variable not set}.sowit.app`)
- traefik.http.routers.flower-ui-https.tls=true
- traefik.http.routers.flower-ui-https.tls.certresolver=le
- traefik.http.routers.flower-ui-http.middlewares=https-redirect
- traefik.http.routers.flower-ui-https.middlewares=traefik-admin-auth
env_file:
- .env
command:
- "--broker=amqp://${RABBITMQ_DEFAULT_USER:-guest}:${RABBITMQ_DEFAULT_PASS:-guest}@rabbitmq:5672//"
depends_on:
- rabbitmq
- redis
networks:
- traefik-public
local_worker:
build:
context: ..
dockerfile: ./setup/devops/docker/app.dockerfile
image: swtools:app
restart: always
volumes:
- ${SWTOOLSWORKINGDIR:-/tmp}:${SWTOOLSWORKINGDIR:-/tmp}
command: ["celery", "--app=app.worker.celery_app:celery_app", "worker", "-n", "local_worker@%h"]
env_file:
- .env
environment:
- RABBITMQ_DEFAULT_USER=${RABBITMQ_DEFAULT_USER}
- RABBITMQ_DEFAULT_PASS=${RABBITMQ_DEFAULT_PASS}
- RABBITMQ_HOST=rabbitmq
- REDIS_HOST=${HOSTNAME?Variable not set}
depends_on:
- rabbitmq
- redis
networks:
- traefik-public
dashboard_app:
image: swtools:app
restart: always
labels:
- traefik.enable=true
- traefik.http.services.dash-app.loadbalancer.server.port=${DASH_PORT-8080}
- traefik.http.routers.dash-app-http.entrypoints=http
- traefik.http.routers.dash-app-http.rule=Host(`dashboard.${HOSTNAME?Variable not set}.sowit.app`)
- traefik.docker.network=traefik-public
- traefik.http.routers.dash-app-https.entrypoints=https
- traefik.http.routers.dash-app-https.rule=Host(`dashboard.${HOSTNAME?Variable not set}.sowit.app`)
- traefik.http.routers.dash-app-https.tls=true
- traefik.http.routers.dash-app-https.tls.certresolver=le
- traefik.http.routers.dash-app-http.middlewares=https-redirect
- traefik.http.middlewares.operator-auth.basicauth.users=${OPERATOR_USERNAME?Variable not set}:${HASHED_OPERATOR_PASSWORD?Variable not set}
- traefik.http.routers.dash-app-https.middlewares=operator-auth
volumes:
- ${SWTOOLSWORKINGDIR:-/tmp}:${SWTOOLSWORKINGDIR:-/tmp}
command: ['waitress-serve', '--port=${DASH_PORT:-8080}', 'app.order_dashboard:app.server']
env_file:
- .env
environment:
- RABBITMQ_DEFAULT_USER=${RABBITMQ_DEFAULT_USER}
- RABBITMQ_DEFAULT_PASS=${RABBITMQ_DEFAULT_PASS}
- RABBITMQ_HOST=rabbitmq
- REDIS_HOST=${HOSTNAME?Variable not set}
networks:
- traefik-public
depends_on:
- rabbitmq
- redis
networks:
traefik-public:
external: true
和我的芹菜配置文件:
import os
import warnings
from pathlib import Path
# result backend use redis
result_backend_host = os.getenv('REDIS_HOST', 'localhost')
result_backend_pass = os.getenv('REDIS_PASS', 'password')
result_backend = 'redis://:{password}@{host}:6379/0'.format(password=result_backend_pass, host=result_backend_host)
# redis_retry_on_timeout = True
redis_socket_keepalive = True
# broker use rabbitmq
rabbitmq_user = os.getenv('RABBITMQ_DEFAULT_USER', 'guest')
rabbitmq_pass = os.getenv('RABBITMQ_DEFAULT_PASS', 'guest')
rabbitmq_host = os.getenv('RABBITMQ_HOST', 'localhost')
broker_url = 'amqp://{user}:{password}@{host}:5672//'.format(user=rabbitmq_user, password=rabbitmq_pass, host=rabbitmq_host)
include = ['app.worker.tasks', 'app.dashboard.example1', 'app.dashboard.example2']
#task events
worker_send_task_events = True
task_send_sent_event = True
所有 env 变量都已定义,除了我的套接字超时问题外,它运行良好!当我在容器实例上部署一个新的工作人员时,我设置了环境变量,以便它连接到在 docker-compose 上运行的 rabbitmq 和 redis。
这是我定义 celery 应用程序的 celery 文件:
from celery import Celery
from app.worker import celery_config
celery_app = Celery()
celery_app.config_from_object(celery_config)
解决方案
最后将后端更改为 rpc 解决了问题。我用 redis 尝试了不同的方法,但没有成功。一种挖掘方法是使用 tcp-dump 检查套接字以查看它阻塞的位置,但我没有尝试使用 rpc 后端解决了我的问题。
推荐阅读
- java - 如何使用对 SD 卡的直接访问权限
- screenshot - SharpDX '不支持这样的接口'
- android - 用牙签在仪器化测试中注入成员
- sql - Maximo SQL BiRT 报告多列 alnvalue
- sql - 唯一约束验证 APEX
- jquery - 如何压缩这些 jQuery 代码?
- java - Spring Reactive 测试用例启动 Netty Server 失败
- redis - Redis 如何获取当前数据库名称
- javascript - 如何访问嵌入式 openload iframe
- javascript - 有没有办法让 HTML 元素溢出:滚动 DIV 始终保持在其他元素之上?