4

下面这两个任务有什么区别？

第一个出错，第二个运行正常。两者都是相同的，它们接受额外的参数并且它们都以相同的方式调用。

ProcessRequests.delay(batch) **error object.__new__() takes no parameters**


SendMessage.delay(message.pk, self.pk) **works!!!!**

现在，我已经意识到错误的含义，但我的困惑是为什么一个有效而另一个无效。

任务...

1) 

class ProcessRequests(Task):
    name = "Request to Process"
    max_retries = 1
    default_retry_delay = 3

    def run(self, batch):
       #do something

2)

class SendMessage(Task):
    name = "Sending SMS"
    max_retries = 10
    default_retry_delay = 3

    def run(self, message_id, gateway_id=None, **kwargs):
        #do something

完整的任务代码....

from celery.task import Task
from celery.decorators import task

import logging

from sms.models import Message, Gateway, Batch
from contacts.models import Contact
from accounts.models import Transaction, Account


class SendMessage(Task):
    name = "Sending SMS"
    max_retries = 10
    default_retry_delay = 3

    def run(self, message_id, gateway_id=None, **kwargs):
        logging.debug("About to send a message.")

        # Because we don't always have control over transactions
        # in our calling code, we will retry up to 10 times, every 3
        # seconds, in order to try to allow for the commit to the database
        # to finish. That gives the server 30 seconds to write all of
        # the data to the database, and finish the view.
        try:
            message = Message.objects.get(pk=message_id)
        except Exception as exc:
            raise SendMessage.retry(exc=exc)


        if not gateway_id:
            if hasattr(message.billee, 'sms_gateway'):
                gateway = message.billee.sms_gateway
            else:
                gateway = Gateway.objects.all()[0]
        else:
            gateway = Gateway.objects.get(pk=gateway_id)

        # Check we have a credits to sent me message
        account = Account.objects.get(user=message.sender)
        # I'm getting the non-cathed version here, check performance!!!!!
        if account._balance() >= message.length:
            response = gateway._send(message)

            if response.status == 'Sent':
                # Take credit from users account.
                transaction = Transaction(
                    account=account,
                    amount=- message.charge,
                    description="Debit: SMS Sent",

                    )
                transaction.save()
                message.billed = True
                message.save()
        else:
            pass


        logging.debug("Done sending message.")


class ProcessRequests(Task):
    name = "Request to Process"
    max_retries = 1
    default_retry_delay = 3

    def run(self, message_batch):
        for e in Contact.objects.filter(contact_owner=message_batch.user, group=message_batch.group):
            msg = Message.objects.create(
                recipient_number=e.mobile,
                content=message_batch.content,
                sender=e.contact_owner,
                billee=message_batch.user,
                sender_name=message_batch.sender_name
            )
            gateway = Gateway.objects.get(pk=2)
            msg.send(gateway)
            #replace('[FIRSTNAME]', e.first_name)

试过：

 ProcessRequests.delay(batch) should work gives error error object.__new__() takes no parameters     
 ProcessRequests().delay(batch) also gives error error object.__new__() takes no parameters
4

1 回答 1

6

我能够重现您的问题：

import celery
from celery.task import Task

@celery.task
class Foo(celery.Task):
    name = "foo"
    def run(self, batch):
       print 'Foo'

class Bar(celery.Task):
    name = "bar"
    def run(self, batch):
       print 'Bar'

# subclass deprecated base Task class
class Bar2(Task):
    name = "bar2"
    def run(self, batch):
       print 'Bar2'

@celery.task(name='def-foo')
def foo(batch):
    print 'foo'

输出：

In [2]: foo.delay('x')
[WARNING/PoolWorker-4] foo

In [3]: Foo().delay('x')
[WARNING/PoolWorker-2] Foo

In [4]: Bar().delay('x')
[WARNING/PoolWorker-3] Bar

In [5]: Foo.delay('x')
TypeError: object.__new__() takes no parameters

In [6]: Bar.delay('x')
TypeError: unbound method delay() must be called with Bar instance as first argument (got str instance instead)

In [7]: Bar2.delay('x')
[WARNING/PoolWorker-1] Bar2

我看到您使用了已弃用celery.task.Task的基类，这就是您没有收到unbound method错误的原因：

Definition: Task(self, *args, **kwargs)
Docstring:
Deprecated Task base class.

Modern applications should use :class:`celery.Task` instead.

我不知道为什么ProcessRequests不起作用。也许是一些缓存问题，您之前可能曾尝试将装饰器应用于您的类并且它被缓存了，这正是您尝试将此装饰器应用于 Task 类时遇到的错误。

删除所有 .pyc 文件，重新启动 celery workers 并重试。

不要直接使用类

  1. 每个（工作）进程只实例化一次任务，因此每次创建任务类的对象（在客户端）没有意义，即Bar()是错误的。
  2. Foo.delay()或Foo().delay()可能或可能不工作，取决于装饰器name参数和类name属性的组合。

celery.registry.tasks从字典中获取任务对象，或者只@celery.task在函数上使用装饰器（foo在我的示例中）。

于 2013-04-25T15:22:02.957 回答