from concurrent.futures.thread import ThreadPoolExecutor


class ThreadPool(object):
  def __init__(self):
    # 線程池
    self.executor = ThreadPoolExecutor(20)
    # 用于存儲(chǔ)每個(gè)項(xiàng)目批量任務(wù)的期程
    self.future_dict = {}

  # 檢查某個(gè)項(xiàng)目是否有正在運(yùn)行的批量任務(wù)
  def is_project_thread_running(self, project_id):
    future = self.future_dict.get(project_id, None)
    if future and future.running():
      # 存在正在運(yùn)行的批量任務(wù)
      return True
    return False

  # 展示所有的異步任務(wù)
  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主線程運(yùn)行的生命周期
global_thread_pool = ThreadPool()

# 檢查異步任務(wù)
if global_thread_pool.is_project_thread_running(project_id):
  raise exceptions.ValidationError(detail='存在正在處理的批量任務(wù)，請稍后重試')

# 提交一個(gè)異步任務(wù)
future = global_thread_pool.executor.submit(self.batch_thread, project_id)
global_thread_pool.future_dict[project_id] = future

# 查看所有異步任務(wù)
@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('異步任務(wù)執(zhí)行失敗:\n %s' % trace_log)
    global_thread_pool.lock.release()

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