Django异步任务线程池实现原理

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

def batch_thread(self):
  for conn in connections.all():
    conn.close_if_unusable_or_obsolete()
  ...