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('異步任務執(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()
  ...