当前位置: 首页 > 知识库问答 >
问题:

工作人员发回芹菜、redis和rabbitMQ结果的延迟时间为2分钟

东郭弘
2023-03-14

我所拥有的:

我用rabbitMQ作为代理运行芹菜,用redis作为结果后端。我有一个发送任务和处理任务的工人的应用程序。

我按如下方式进行了部署:

  • 应用程序、redis、RabbitMQ和一个工作人员(让我们称他为“local_worker”)正在使用docker-compose在azure VM上运行,因此我使用docker版本的RabbitMQ和redis(6.2.5)。RabbitMQ和redis端口在VM上打开,这些容器配置了用户名和密码。
  • 我使用连接到VM上运行的redis和RabbitMQ的azure容器实例添加工作人员。

首先,如果你对这个架构有建议,我很乐意得到建议。

问题:

一切都很好,任务被分派给不同的工作人员,这些工作人员返回结果等。。。

如果在30分钟后没有任务运行的情况下发送任务,我观察到redis延迟为2分钟,此时任务没有发送到“local_worker”。

  • 我知道这一定来自redis,因为我可以在发送任务后立即在worker容器实例中看到任务的日志
  • 我使用flower和graphana以及celery prometheus exporter来监控此架构,以便监控任务的延迟。在flower上,潜在任务处于“加工”状态
  • 对于无任务间隔后的第一个任务,且“local_worker”未处理的任务,还有120秒的时间
  • 当任务由与redis在同一个VM上运行的“local_worker”处理时,不会发生这种情况

这就像redis或VM在发回结果之前沉睡了2分钟。因为正好是120秒(2分钟),我希望它是redis、芹菜或azure想要的东西(确定性的东西)

我不使用redis.conf文件,只使用默认设置(密码除外)来运行redis服务器。

感谢您对我的架构和问题的帮助和反馈。

第一个和第三个任务已经由本地工作人员处理。第二个是由外部工作人员处理的。在外部工作人员的日志上,我在返回结果之前放了一条打印线,这条线是在14:14:23打印的。所以从这次打印到任务的正式结束有120秒。

编辑:

我发现redis_socket_timeout的默认值为120秒。

我删除了行redis_retry_on_timeout=True并在我的芹菜配置文件中添加了行redis_socket_keepalive=True。现在我得到的错误是任务失败,redis.exceptions.TimeoutError:从套接字读取超时。我不知道为什么套接字超时而结果已准备好。是我的容器实例的网络问题吗?

这是我的docker-compose:

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}.example.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}.example.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}.example.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}.example.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-html" target="_blank">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}.example.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}.example.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变量都已定义,除了我的套接字超时问题外,它工作得很好!当我在容器实例上部署新的工作线程时,我设置了 env 变量,以便它连接到在 docker 组合上运行的 rabbitmq 和 redis。

下面是我的芹菜文件,它定义了芹菜应用程序:

from celery import Celery
from app.worker import celery_config

celery_app = Celery()
celery_app.config_from_object(celery_config)

共有2个答案

阳兴朝
2023-03-14

最后,将后端更改为rpc解决了问题。我用redis尝试了不同的方法,但都不起作用。挖掘的一种方法是使用tcp转储检查套接字,以查看其阻塞的位置,但我没有尝试,因为我的问题是通过rpc后端解决的。

郑声
2023-03-14

我猜您的Redis实例和工作人员之间有一些防火墙。您可以登录到那个SandboxHost...并确保您可以连接您的redis吗?

您可以使用telnet执行此操作,例如:

telnet <your_redis_hostname> <your_redis_port>

或者使用 redis-cli:

redis-cli -h <your_redis_hostname> -p <your_redis_port>

编辑:

看来你错过了result_backend:

result_backend = f"redis://username:{result_backend_pass}@{result_backend_host}:6379/0"

并确保您的REDIS_HOST=${主机名?变量未设置} 是否有效...

编辑2:

可以在Redis命令中添加< code>bind吗:

["redis-server", "--bind", "0.0.0.0", "--requirepass", "${RABBITMQ_DEFAULT_PASS:-password}"]

请注意其安全影响!

 类似资料:
  • 问题内容: 在我的配置文件中,我设置了: 据我所知,守护进程生成8组芹菜的工人,但我完全不知道该怎么和一起做。我认为并发是一种指定工作人员可以使用的最大线程数的方法,而自动伸缩是工作人员在必要时扩展和缩减子级工作人员的一种方法。 这些任务的负载比较大(大约20-50kB),大约有2-3百万个这样的任务,但是每个任务的运行时间不到一秒钟。我看到内存使用量激增,因为代理将任务分配给每个工作人员,从而多

  • 我是芹菜、Redis和RabbitMQ的新手。 目前,我正在使用RabbitMQ作为消息代理,并且在配置中没有设置任何内容。(使用Django、MySQL) 我想知道是否有可能使用Redis作为后端的结果存储,同时将RabbitMQ用作消息代理。 我知道的只是添加一些设置,

  • 我创建了一个应用程序,其中一个地点列表保存在数据库中。我只根据Google留档保存了保存地点的地点ID。 然后我创建了一个界面,让用户输入他们的当前位置(mCurrentLocation)。 然后,我创建了一个异步任务,检查数据库中保存的每个地点ID,并检查哪些地点距离用户位置500米以内。这是在doInBackground中完成的。根据Google文档,我使用了getPlaceById和setR

  • 我经常看到两个参与者之间有很长的延迟(60+秒),从第一个参与者发送消息到第二个参与者,以及第二个参与者的方法随消息实际调用时。我可以寻找哪些类型的东西来调试这个问题? ActorA的每个实例都使用为ActorB发送一条消息。在ActorA中调用方法并在ActorB的开始处获得另一个时间戳之后,我立即收集了一个毫秒时间戳(使用)。这些时间戳之间的间隔一致为60秒或更长。具体地说,当按时间绘制时,该

  • 在我的控制器中,我返回由另一个线程填充的延迟结果: FooController.java 页面准备者.java 显然,再次调用所有筛选器以“完成”请求。我的一个过滤器,需要访问 豆,这是请求范围的。该调用引发异常 查看我配置的异常: 但是没有区别。我该怎么解决这个问题?

  • 我最近开始研究分布式计算以提高计算速度。我选择了芹菜。然而,我对一些术语不太熟悉。所以,我有几个相关的问题。 来自芹菜文档: ... Celery通过消息进行通信,通常使用代理在客户机和工作人员之间进行调解。为了启动任务,客户机将消息添加到队列中,然后代理将该消息传递给工作者。 什么是客户端(这里)?什么是经纪商?为什么消息通过代理传递?为什么 Celery 会使用后端和队列进行进程间通信? 当我