多线程消费， 队列没有清空就退出了， 请教原因， 谢谢！

[vegetableChick]

我想通过 ThreadPoolExecutor 使用多个线程来消耗 redis 队列。但进程在队列没有消耗完的情况下退出了

下面是实现代码

from concurrent.futures import ThreadPoolExecutor

import redis
from redis import Redis



pool = redis.ConnectionPool(
    max_connections=settings.REDIS_POOL_MAX_CLIENT,
    host=settings.REDIS_HOST,
    port=settings.REDIS_PORT,
    db=8,
    decode_responses=True
)

redis_con = Redis(connection_pool=pool)


class BasicTask(object):
     def __init__(self,
                 consume_queue_name=None,
                 thread_num=50):
        self.consume_queue_name = consume_queue_name
        self.thread_num = thread_num

    def _consume(self):
        try:
            with ThreadPoolExecutor(max_workers=self.thread_num) as e:
                e.map(self._do_request, range(0, self.thread_num))
        except Exception as e:
            self.logger.error(f"[consume error]: {e}")

    def _do_request(self, _):
        try:
            with redis_con as redis_conn:
                while 1:
                    account_id_info = redis_conn.rpop(self.consume_queue_name)
                    if account_id_info:
                        try:
                            # django orm save db
                            ...

                        except Exception as e:
                            import traceback as tb
                            tb.print_exc()
                            self.logger.error(f"[consume error]: {e}. ")

                    else:
                        break

        except Exception as e:
            self.logger.error(f"[Unexpected Error: {e}]")
            import traceback as tb
            tb.print_exc()

    def run(self):
        self._consume()



# run 
BasicTask(consume_queue_name="base_list_queue").run()

请问 bug 写在哪里了？ 感谢大佬

python3.7.3

[MoYi123]

唯一的可能性就是 if account_id_info: 后面的 break 了吧, 不然都是有日志的.

[wuwukai007]

不是应该 brpop 之后在开启多线程 执行任务吗，怎么多线程 brpop 了