标签: celery

app = Celery('myapp',
             broker='amqp://user:pass#1@localhost:5672//',
             backend='rpc://',
             include=['myapp.tasks'])

我收到这个错误

ValueError：以 10 为基数的 int() 的文字无效：“pass”

这段代码不起作用，我是 python 新手，Django 是否有转义序列？

我尝试了 u"", r"" ,'#' , '##' 和 '#' ，希望它能逃脱它，但它没有。

Traceback (most recent call last):
  File "C:\Program Files (x86)\Microsoft Visual Studio\Shared\Python36_64\lib\runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "C:\Program Files (x86)\Microsoft Visual Studio\Shared\Python36_64\lib\runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "C:\Users\user\source\repos\BtcApi\BtcApi\btcapienv\Scripts\celery.exe\__main__.py", line 9, in <module>
  File "c:\users\user\source\repos\btcapi\btcapi\btcapienv\lib\site-packages\celery\__main__.py", line 14, in main
    _main()
  File "c:\users\user\source\repos\btcapi\btcapi\btcapienv\lib\site-packages\celery\bin\celery.py", line 326, in main
    cmd.execute_from_commandline(argv)
  File "c:\users\user\source\repos\btcapi\btcapi\btcapienv\lib\site-packages\celery\bin\celery.py", line 488, in execute_from_commandline …

我正在使用celery和rabbitmq，但是由于在队列中推送了多个任务，我的服务器内存利用率变得超过40%，因此rabbit将不再接受任何任务。所以我想删除那些已经执行的消息，但是由于rabbitmq的持久行为，这些消息不会自动删除，所以我想设置一些配置，例如 autoAck=True ，这样如果从 celery 消耗消息，它将从rabbitmq 队列以及我的服务器内存。请解释一下我们该如何做到这一点。

在进行任何解释之前，这是我的项目的树

| projectname
|____|__init__.py
|____|celery.py
|____|settings.py
|____|urls.py
|____|wsgi.py
|app1
|app2

这是我的芹菜.py

from celery import Celery
from celery import shared_task

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'projectname.settings')
app = Celery('projectname')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

from app1.models import *

@share_task
def tasks():
     ''' '''

每次我尝试使用此行导入文件时，models我都会得到：celery.pyfrom app1.models import *

django.core.exceptions.AppRegistryNotReady：应用程序尚未加载。

并且本地服务器突然停止工作。 这篇文章涉及类似的问题，但不确定这里是否是这种情况。

我想要的是将一些模型导入到文件中，这样我就可以使用它们进行一些查询。

我对可能出现的问题有一点线索，但不确定。

viewsimport stuff from import stuff from就像要执行的任务尝试从 导入东西。models.py
viewscelery.py
celery.pymodels

所以那个像蛇咬自己尾巴的圆圈对我来说很奇怪。

撤销 @periodic_task 发送的任务Discarding revoked tasks & Due task to workers.

芹菜工人屏幕截图

[2018-09-17 12:23:50,864: INFO/MainProcess] Received task: cimexapp.tasks.add[xxxxxxx]
[2018-09-17 12:23:50,864: INFO/MainProcess] Discarding revoked task: cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:00,865: INFO/Beat] Scheduler: Sending due task cimexapp.tasks.add (cimexapp.tasks.add) [2018-09-17 12:24:00,869: INFO/MainProcess] Received task: cimexapp.tasks.add[xxxxxxx]
[2018-09-17 12:24:00,869: INFO/MainProcess] Discarding revoked task: cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:10,865: INFO/Beat] Scheduler: Sending due task cimexapp.tasks.add (cimexapp.tasks.add) [2018-09-17 12:24:10,868: INFO/MainProcess] Received task: cimexapp.tasks.add[xxxxxxx]
[2018-09-17 12:24:10,869: INFO/MainProcess] Discarding revoked task: cimexapp.tasks.add[xxxxxxx]

任务.py

@periodic_task(run_every=timedelta(seconds=10),options={"task_id":"xxxxxxx"})
def add():
     call(["ping","-c10","google.com"])


def stop():
    x = revoke("xxxxxxx",terminate=True,signal="KILL")
    print(x) …

我已经阅读了类似的主题并完成了建议的所有内容，但问题仍然存在。

我正在将我的应用程序部署到 Heroku。本地一切工作正常，但在指定每个设置后的部署过程中，我可以考虑指定 celery 工作人员发送以下错误：

：22：50.722826 + 00：00应用程序[worker.1]：[2018-10-21 18：22：50,722：错误/MainProcess]消费者：无法连接到amqp://guest：**@127.0.0.1:5672 //: [Errno 111] 连接被拒绝。

我尝试从 CloudAMQP 切换到 Redis。问题仍然存在。

这是我的配置文件：

Django的settings.py：

try: 
    CELERY_BROKER_URL = os.environ['REDIS_URL']
except KeyError: 
    CELERY_BROKER_URL = 'amqp://admin:admin@localhost:5672/admin_host'

try:
    CELERY_RESULT_BACKEND = os.environ['REDIS_URL']
except KeyError:
    CELERY_RESULT_BACKEND = 'amqp://admin:admin@localhost:5672/admin_host'


CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

django_heroku.settings(locals())

芹菜.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from . import settings

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'file_upload.settings')

app = Celery('file_upload', broker_pool_limit=1) …

我想在 digitalocean 上使用 docker 部署我的小 Web 应用程序。问题是，我没有分开 Flask + celery，所以所有东西都在一个容器中。\n每次我运行docker-compose upWeb 容器都会崩溃（“以代码 0 退出”）。

项目结构如下：

.\n.\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 docker-compose.yml\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 nginx\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 dockerfile\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 nginx.conf\n\xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 web\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 app.ini\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 bot\n    \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n    \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 main.py\n    \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __pycache__\n    \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 timer.py\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 celeryd.pid\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 config.py\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 dockerfile\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 dockerignore\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 flask_app.py\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 requirements.txt\n    \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 tasks.py\n    \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 webapp\n        \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 bot\n        \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n        \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 models.py\n        \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __pycache__\n        \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 static\n        \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 templates\n        \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 users\n\n

这是 docker-compose.yml：

.\n.\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 …

我曾经docker-compose创建过一个包含 redis 和 celery 的 docker 实例。这将成功构建，但是，在我的 python 脚本中，当我尝试将某些内容添加到队列中时，出现以下错误：

Error 8 connecting to redis:6379. nodename nor servname provided, or not known.

我尝试更改我的 celeryconfig 文件地址，并尝试了此处建议的解决方案： https: //github.com/mozilla-services/cliquet/issues/664但我无法修复它。

还有其他人遇到过这样的问题吗？

** example.py 的一部分 **

app = Celery('server', broker='redis://redis:6379/0')
app.config_from_object(celeryconfig)

@app.task(trail=True)
def count_words_at_url(url):
    time.sleep(50)
    resp = requests.get(url)
    return len(resp.text.split())

celeryconfig.py

result_backend = 'redis://redis:6379/0'
broker_url = 'redis://redis:6379/0'
worker_prefetch_multiplier = 1
worker_concurrency  =1

Dockerfile.celery

FROM python:3.6-alpine
WORKDIR /usr/src/app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ulimit -n …

我有正在运行的 Celery 3.1 应用程序，它记录了一些敏感信息。理想情况下，我希望有相同的日志，但没有结果部分。

目前它看起来像：

worker_1 | [2019-12-10 13:46:40,052: INFO/MainProcess] Task xxxxx succeeded in 13.19569299298746s: yyyyyyy

我想拥有：

worker_1 | [2019-12-10 13:46:40,052: INFO/MainProcess] Task xxxxx succeeded in 13.19569299298746s

怎么做？

编辑：看来这可以完成这项工作：https://docs.celeryproject.org/en/3.1/reference/celery.worker.job.html#celery.worker.job.Request.success_msg但我不知道如何来实际使用它。

我在 django 应用程序中使用 celery。如果发生故障，我会引发一个自定义异常，该异常是用此类定义的：

class CustomException(Exception):
    def __init__(self, err_input, err_info=''):
        self.err_key, self.err_msg, self.app_domain = err_input
        self.message = self.err_msg
        super(CustomException, self).__init__(self.message)
        self.err_info = err_info

    def __str__(self):
        return '[{}] {}'.format(str(self.err_key), str(self.err_msg))

    def __repr__(self):
        return self.message

用于error_info内部日志记录目的。如果我抛出在相关 celery 任务中收到的消息：

{
  "task_id": "0dfd5ef3-0df1-11eb-b74a-0242ac110002",
  "status": "FAILURE",
  "result": {
    "exc_type": "MaybeEncodingError",
    "exc_message": [
      "'PicklingError(\"Can\\'t pickle <class \\'module.CustomException\\'>: it\\'s not the same object as module.CustomException\")'",
      "\"(1, <ExceptionInfo: CustomException('Error message')>, None)\""
    ],
    "exc_module": "billiard.pool"
  },
  "traceback": null,
  "children": []
}

我在异常类中配置了什么错误？

我正在尝试将 Celery 添加到 django 来安排任务。我使用 Redis 后端，并通过 unix 套接字连接。安装程序一直有效，直到我尝试使用密码身份验证到 redis.conf：

我的设置.py：

CELERY_BROKER_URL = 'redis+socket:///home/username/domain/redis.sock?password=mypasswd'
CELERY_RESULT_BACKEND = 'redis+socket:///home/username/domain/redis.sock?password=mypasswd'

芹菜.py：

from __future__ import absolute_import, unicode_literals

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'public_python.settings')
# celery settings for the demo_project
app = Celery('public_python')
app.config_from_object('django.conf:settings', namespace='CELERY')

app.autodiscover_tasks()

结果：

[2021-07-19 21:22:14,003: ERROR/MainProcess] consumer: Cannot connect to redis+socket:///home/username/domain/redis.sock: Authentication required..

我已经尝试添加：（ CELERY_REDIS_PASSWORD='mypasswd' 类似单词的任何可能的组合），但没有运气。