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()
    ...