django执行阻塞任务怎么处理(django定时触发任务)

导读：本篇文章首席CTO笔记来给大家介绍有关django执行阻塞任务怎么处理的相关内容，希望对大家有所帮助，一起来看看吧。

Django配置Celery执行异步和同步任务（tasks））

celery是一个基于python开发的简单、灵活且可靠的分布式任务队列框架，支持使用任务队列的方式在分布式的机器/进程/线程上执行任务调度。采用典型的生产者-消费者模型，主要由三部分组成：

比如系统上线前后台批量导入历史数据，发送短信、发送邮件等耗时的任务

1.安装RabbitMQ，这里我们使用RabbitMQ作为broker，安装完成后默认启动了，也不需要其他任何配置

Ubuntu linux安装

CentOS Linux 安装

苹果mac 安装需要配置

配置环境变量 (苹果用户)

启动rabbitmq-server

2.安装celery

3.celery用在django项目中，django项目目录结构(简化)如下

4.创建 oa/celery.py 主文件

5.在 oa/__init__.py 文件中增加如下内容，确保django启动的时候这个app能够被加载到

6.各应用创建tasks.py文件，这里为 users/tasks.py

7.views.py中引用使用这个tasks异步处理

8.启动celery

9.这样在调用post这个方法时，里边的add就可以异步处理了

定时任务的使用场景就很普遍了，比如我需要定时发送报告给老板~

1. oa/celery.py 文件添加如下配置以支持定时任务crontab

3.启动celery beat，celery启动了一个beat进程一直在不断的判断是否有任务需要执行

Django中怎么使用django-celery完成异步任务

许多Django应用需要执行异步任务, 以便不耽误http request的执行. 我们也可以选择许多方法来完成异步任务, 使用Celery是一个比较好的选择, 因为Celery

有着大量的社区支持, 能够完美的扩展, 和Django结合的也很好. Celery不仅能在Django中使用, 还能在其他地方被大量的使用. 因此一旦学会使用Celery, 我

们可以很方便的在其他项目中使用它.

1. Celery版本

本篇博文主要针对Celery 3.0.x. 早期版本的Celery可能有细微的差别.

2. Celery介绍

Celery的主要用处是执行异步任务, 可以选择延期或定时执行功能. 为什么需要执行异步任务呢?

第一, 假设用户正发起一个request, 并等待request完成后返回. 在这一request后面的view功能中, 我们可能需要执行一段花费很长时间的程序任务, 这一时间

可能远远大于用户能忍受的范围. 当这一任务并不需要立刻执行时, 我们便可以使用Celery在后台执行, 而不影响用户浏览网页. 当有任务需要访问远程服务器完

成时, 我们往往都无法确定需要花费的时间.

第二则是定期执行某些任务. 比如每小时需要检查一下天气预报, 然后将数据储存到数据库中. 我们可以编写这一任务, 然后让Celery每小时执行一次. 这样我们

的web应用便能获取最新的天气预报信息.

我们这里所讲的任务task, 就是一个Python功能(function). 定期执行一个任务可以被认为是延时执行该功能. 我们可以使用Celery延迟5分钟调用function

task1, 并传入参数(1, 2, 3). 或者我们也可以每天午夜运行该function.

我们偏向于将Celery放入项目中, 便于task访问统一数据库和Django设置.

当task准备运行时, Celery会将其放入列队queue中. queue中储存着可以运行的task的list. 我们可以使用多个queue, 但为了简单, 这里我们只使用一个.

将任务task放入queue就像加入todo list一样. 为了使task运行, 我们还需要在其他线程中运行的苦工worker. worker实时观察着代运行的task, 并逐一运行这

些task. 你可以使用多个worker, 通常他们位于不同服务器上. 同样为了简单起见, 我们这只是用一个worker.

我们稍后会讨论queue, worker和另外一个十分重要的进程, 接下来我们来动动手:

3. 安装Celery

我们可以使用pip在vietualenv中安装:

pip install django-celery

4. Django设置

我们暂时使用django runserver来启动celery. 而Celery代理人(broker), 我们使用Django database broker implementation. 现在我们只需要知道Celery

需要broker, 使用django自身便可以充当broker. (但在部署时, 我们最好使用更稳定和高效的broker, 例如Redis.)

在settings.py中:

import djcelery

djcelery.setup_loader()

BROKER_URL = 'django://'

...

INSTALLED_APPS = (

...

'djcelery',

'kombu.transport.django',

...

)

第一二项是必须的, 第三项则告诉Celery使用Django项目作为broker.

在INSTALLED_APPS中添加的djcelery是必须的. kombu.transport.django则是基于Django的broker

最后创建Celery所需的数据表, 如果使用South作为数据迁移工具, 则运行:

python manage.py migrate

否则运行: (Django 1.6或Django 1.7都可以)

python manage.py syncdb

5. 创建一个task

正如前面所说的, 一个task就是一个Pyhton function. 但Celery需要知道这一function是task, 因此我们可以使用celery自带的装饰器decorator: @task. 在

django app目录中创建taske.py:

from celery import task

@task()

def add(x, y):

return x + y

当settings.py中的djcelery.setup_loader()运行时, Celery便会查看所有INSTALLED_APPS中app目录中的tasks.py文件, 找到标记为task的function, 并

将它们注册为celery task.

将function标注为task并不会妨碍他们的正常执行. 你还是可以像平时那样调用它: z = add(1, 2).

6. 执行task

让我们以一个简单的例子作为开始. 例如我们希望在用户发出request后异步执行该task, 马上返回response, 从而不阻塞该request, 使用户有一个流畅的访问

过程. 那么, 我们可以使用.delay, 例如在在views.py的一个view中:

from myapp.tasks import add

...

add.delay(2, 2)

...

Celery会将task加入到queue中, 并马上返回. 而在一旁待命的worker看到该task后, 便会按照设定执行它, 并将他从queue中移除. 而worker则会执行以下代

码:

import myapp.tasks.add

myapp.tasks.add(2, 2)

7. 关于import

这里需要注意的是, 在impprt task时, 需要保持一致. 因为在执行djcelery.setup_loader()时, task是以INSTALLED_APPS中的app名,

加.tasks.function_name注册的, 如果我们由于python path不同而使用不同的引用方式时(例如在tasks.py中使用from myproject.myapp.tasks import

add形式), Celery将无法得知这是同一task, 因此可能会引起奇怪的bug.

8. 测试

a. 启动worker

正如之前说到的, 我们需要worker来执行task. 以下是在开发环境中的如何启动worker:

首先启动terminal, 如同开发django项目一样, 激活virtualenv, 切换到django项目目录. 然后启动django自带web服务器: python manage.py runserver.

然后启动worker:

python manage.py celery worker --loglevel=info

此时, worker将会在该terminal中运行, 并显示输出结果.

b. 启动task

打开新的terminal, 激活virtualenv, 并切换到django项目目录:

$ python manage.py shell

from myapp.tasks import add

add.delay(2, 2)

此时, 你可以在worker窗口中看到worker执行该task:

[2014-10-07 08:47:08,076: INFO/MainProcess] Got task from broker: myapp.tasks.add[e080e047-b2a2-43a7-af74-d7d9d98b02fc]

[2014-10-07 08:47:08,299: INFO/MainProcess] Task myapp.tasks.add[e080e047-b2a2-43a7-af74-d7d9d98b02fc] succeeded in 0.183349132538s: 4

9. 另一个例子

下面我们来看一个更为真实的例子, 在views.py和tasks.py中:

# views.py

from myapp.tasks import do_something_with_form_data

def view(request):

form = SomeForm(request.POST)

if form.is_valid():

data = form.cleaned_data

# Schedule a task to process the data later

do_something_with_form_data.delay(data)

return render_to_response(...)

# tasks.py

@task

def do_something_with_form_data(data):

call_slow_web_service(data['user'], data['text'], ...)

10. 调试

由于Celery的运行需要启动多个部件, 我们可能会漏掉一两个. 所以我们建议:

使用最简单的设置

使用python debug和logging功能显示当前的进程

11. Eager模式

如果在settings.py设置:

CELERY_ALWAYS_EAGER = True

那么Celery便以eager模式运行, 则task便不需要加delay运行:

# 若启用eager模式, 则以下两行代码相同

add.delay(2, 2)

add(2, 2)

12. 查看queue

因为我们使用了django作为broker, queue储存在django的数据库中. 这就意味着我们可以通过django admin查看该queue:

# admin.py

from django.contrib import admin

from kombu.transport.django import models as kombu_models

admin.site.register(kombu_models.Message)

13. 检查结果

每次运行异步task后, Celery都会返回AsyncResult对象作为结果. 你可以将其保存, 然后在将来查看该task是否运行成功和返回结果:

# views.py

result = add.delay(2, 2)

...

if result.ready():

print "Task has run"

if result.successful():

print "Result was: %s" % result.result

else:

if isinstance(result.result, Exception):

print "Task failed due to raising an exception"

raise result.result

else:

print "Task failed without raising exception"

else:

print "Task has not yet run"

14. 定期任务

还有一种Celery的常用模式便是执行定期任务. 执行定期任务时, Celery会通过celerybeat进程来完成. Celerybeat会保持运行, 一旦到了某一定期任务需要执

行时, Celerybeat便将其加入到queue中. 不像worker进程, Celerybeat只有需要一个即可.

启动Celerybeat:

python manage.py celery beat

使Celery运行定期任务的方式有很多种, 我们先看第一种, 将定期任务储存在django数据库中. 即使是在django和celery都运行的状态, 这一方式也可以让我们

方便的修改定期任务. 我们只需要设置settings.py中的一项便能开启这一方式:

# settings.py

CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'

django的异步请求非阻塞是怎么实现的

你应该是使用了Django自己的开发服务器跑的例子，在Django关于manage.py的文档中写道：

--nothreading

The development server is multithreaded by default. Use the --nothreading option to disable the use of threading in the development server.

也就是说，默认情况下你使用./manage.py runserver会开启多个线程对HTTP请求进行伺服，所以第二个请求进来时虽然第一个请求仍在sleep，但已经新开了一个线程进行响应处理，看起来像是“非阻塞”的工作模式，其实质是多线程而非单线程，想禁用这一行为也已经给出了答案，加上--nothreading参数：./manage.py runserver --nothreading即可。

线程池，每个线程都阻塞了怎么办

创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程。在任意点，在大多数 nThreads 线程会处于处理任务的活动状态。如果在所有线程处于活动状态时提交附加任务，则在有可用线程之前，附加任务将在队列中等待。如果在关闭前的执行期间由于失败而导致任何线程终止，那么一个新线程将代替它执行后续的任务（如果需要）。在某个线程被显式地关闭之前，池中的线程将一直存在。

怎么解决阻塞太久线程自动退出

终止线程的三种方法：

1. 使用退出标志，使线程正常退出，也就是当run方法完成后线程终止。

2. 使用stop方法强行终止线程（这个方法不推荐使用，因为stop和suspend、resume一样，也可能发生不可预料的结果）。

3. 使用interrupt方法中断线程。

使用退出标志终止线程：

当run方法执行完后，线程就会退出。但有时run方法是永远不会结束的。如在服务端程序中使用线程进行监听客户端请求，或是其他的需要循环处理的任务。在这种情况下，一般是将这些任务放在一个循环中，如while循环。如果想让循环永远运行下去，可以使用while（true）{……}来处理。但要想使while循环在某一特定条件下退出，最直接的方法就是设一个boolean类型的标志，并通过设置这个标志为true或false来控制while循环是否退出。下面给出了一个利用退出标志终止线程的例子。

package chapter2;

public class ThreadFlag extends Thread

{

public volatile boolean exit = false;

public void run()

{

while (!exit);

}

public static void main(String[] args) throws Exception

{

ThreadFlag thread = new ThreadFlag();

thread.start();

sleep(5000); // 主线程延迟5秒

thread.exit = true; // 终止线程thread

thread.join();

System.out.println("线程退出!");

}

在上面代码中定义了一个退出标志exit，当exit为true时，while循环退出，exit的默认值为false.在定义exit时，使用了一个Java关键字volatile，这个关键字的目的是使exit同步，也就是说在同一时刻只能由一个线程来修改exit的值，

2. 使用stop方法终止线程

使用stop方法可以强行终止正在运行或挂起的线程。我们可以使用如下的代码来终止线程：

thread.stop();

虽然使用上面的代码可以终止线程，但使用stop方法是很危险的，就象突然关闭计算机电源，而不是按正常程序关机一样，可能会产生不可预料的结果，因此，并不推荐使用stop方法来终止线程。

3. 使用interrupt方法终止线程

使用interrupt方法来终端线程可分为两种情况：

（1）线程处于阻塞状态，如使用了sleep方法。

（2）使用while（！isInterrupted（））{……}来判断线程是否被中断。

在第一种情况下使用interrupt方法，sleep方法将抛出一个InterruptedException例外，而在第二种情况下线程将直接退出。下面的代码演示了在第一种情况下使用interrupt方法。

package chapter2;

public class ThreadInterrupt extends Thread

{

public void run()

{

try

{

sleep(50000); // 延迟50秒

}

catch (InterruptedException e)

{

System.out.println(e.getMessage());

}

public static void main(String[] args) throws Exception

{

Thread thread = new ThreadInterrupt();

thread.start();

System.out.println("在50秒之内按任意键中断线程!");

System.in.read();

thread.interrupt();

thread.join();

System.out.println("线程已经退出!");

}

上面代码的运行结果如下：

在50秒之内按任意键中断线程!

sleep interrupted

线程已经退出!

在调用interrupt方法后， sleep方法抛出异常，然后输出错误信息：sleep interrupted.

注意：在Thread类中有两个方法可以判断线程是否通过interrupt方法被终止。一个是静态的方法interrupted（），一个是非静态的方法isInterrupted（），这两个方法的区别是interrupted用来判断当前线是否被中断，而isInterrupted可以用来判断其他线程是否被中断。因此，while （！isInterrupted（））也可以换成while （！Thread.interrupted（））。