这篇文章主要介绍了Django异步任务线程池实现原理,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
当数据库数据量很大时(百万级),许多批量数据修改请求的响应会非常慢,一些不需要即时响应的任务可以放到后台的异步线程中完成,发起异步任务的请求就可以立即响应
选择用线程池的原因是:线程比进程更为可控。不像子进程,子线程会在所属进程结束时立即结束。线程可共享内存。
请求任务异步处理的原理
使用python manage.py runserver模式启动的Django应用只有一个进程,对于每个请求,主线程会开启一个子线程来处理请求。请求子线程向主线程申请一个新线程,然后把耗时的任务交给新线程,自身立即响应,这就是请求任务异步处理的原理。
可视化线程池
如果想要管理这批异步线程,知道他们是否在运行中,可以使用线程池(ThreadPoolExecutor)。
线程池会先启动若干数量的线程,并让这些线程都处于睡眠状态,当向线程池submit一个任务后,会唤醒线程池中的某一个睡眠线程,让它来处理这个任务,当处理完这个任务,线程又处于睡眠状态。
submit任务后会返回一个期程(future),这个对象可以查看线程池中执行此任务的线程是否仍在处理中
因此可以构建一个全局可视化线程池:
from concurrent.futures.thread import ThreadPoolExecutor class ThreadPool(object): def __init__(self): # 线程池 self.executor = ThreadPoolExecutor(20) # 用于存储每个项目批量任务的期程 self.future_dict = {} # 检查某个项目是否有正在运行的批量任务 def is_project_thread_running(self, project_id): future = self.future_dict.get(project_id, None) if future and future.running(): # 存在正在运行的批量任务 return True return False # 展示所有的异步任务 def check_future(self): data = {} for project_id, future in self.future_dict.items(): data[project_id] = future.running() return data def __del__(self): self.executor.shutdown() # 主线程中的全局线程池 # global_thread_pool的生命周期是Django主线程运行的生命周期 global_thread_pool = ThreadPool()
使用:
# 检查异步任务 if global_thread_pool.is_project_thread_running(project_id): raise exceptions.ValidationError(detail='存在正在处理的批量任务,请稍后重试') # 提交一个异步任务 future = global_thread_pool.executor.submit(self.batch_thread, project_id) global_thread_pool.future_dict[project_id] = future # 查看所有异步任务 @login_required def check_future(request): data = global_thread_pool.check_future() return HttpResponse(status=status.HTTP_200_OK, content=json.dumps(data))
串行执行
使用线程锁
在全局线程池中初始化线程锁
class ThreadPool(object): def __init__(self): self.executor = ThreadPoolExecutor(20) self.future_dict = {} self.lock = threading.Lock()
然后执行线程前需要获取锁并再执行结束后释放锁
def batch_thread(self): global_thread_pool.lock.acquire() try: ... global_thread_pool.lock.release() except Exception: trace_log = traceback.format_exc() logger.error('异步任务执行失败:\n %s' % trace_log) global_thread_pool.lock.release()
需要捕捉异常预防子线程出错而无法释放锁的情况
异步线程任务执行前先检查数据库连接是否可用,然后关掉不可用连接
由于django的数据库连接是保存到线程本地变量中的,通过ThreadPoolExecutor创建的线程会保存各自的数据库连接。
当连接被保存的时间超过mysql连接的最大超时时间,连接失效,但不会被线程释放。
之后再调起线程执行涉及到数据库操作的异步任务时,会用到失效的数据库连接,导致报错“MySQL server has gone away”。
解决方案是在线程池的所有异步任务执行前先检查数据库连接是否可用,然后关掉不可用连接
def batch_thread(self): for conn in connections.all(): conn.close_if_unusable_or_obsolete() ...
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持小牛知识库。
我试图在这个类中实现异步任务,但问题是我在我的程序中调用了函数,该函数返回一个值,我不知道该把它放在哪里。在异步任务中,我应该在哪里定义?我得到以下例外 以下是我的主要课程: 这是我的解析类:公共类解析{ List headlines列出链接;列表描述;列出lstDate列出新日期;//字符串a,b,c,d;public InputStream getInputStream(URL URL){ t
本文向大家介绍Spring Boot使用Spring的异步线程池的实现,包括了Spring Boot使用Spring的异步线程池的实现的使用技巧和注意事项,需要的朋友参考一下 前言 线程池,从名字上来看,就是一个保存线程的"池子",凡事都有其道理,那线程池的好处在哪里呢? 我们要让计算机为我们干一些活,其实都是在使用线程,使用方法就是new一个Runnable接口或者新建一个子类,继承于Threa
问: 如何异步处理繁重的业务,避免主业务被长时间阻塞。例如我要给1000用户发送邮件,这个过程很慢,可能要阻塞数秒,这个过程中因为主流程被阻塞,会影响后续的请求,如何将这样的繁重任务交给其它进程异步处理。 答: 可以在本机或者其它服务器甚至服务器集群预先建立一些任务进程处理繁重的业务,任务进程数可以开多一些,例如cpu的10倍,然后调用方利用AsyncTcpConnection将数据异步发送给这些
我在StackOverflow的职业生涯中有一个第一个问题。我希望是你帮我。 发生了什么:
本文向大家介绍Java线程池FutureTask实现原理详解,包括了Java线程池FutureTask实现原理详解的使用技巧和注意事项,需要的朋友参考一下 前言 线程池可以并发执行多个任务,有些时候,我们可能想要跟踪任务的执行结果,甚至在一定时间内,如果任务没有执行完成,我们可能还想要取消任务的执行,为了支持这一特性,ThreadPoolExecutor提供了 FutureTask 用于追踪任务的
本文向大家介绍Java多线程之异步Future机制的原理和实现,包括了Java多线程之异步Future机制的原理和实现的使用技巧和注意事项,需要的朋友参考一下 项目中经常有些任务需要异步(提交到线程池中)去执行,而主线程往往需要知道异步执行产生的结果,这时我们要怎么做呢?用runnable是无法实现的,我们需要用callable看下面的代码: 虽然可以实现获取异步执行结果的需求,但是我们发现这个F