python-asyncio - asyncio.create_connection 在每次调用时实例化一个 ThreadPoolExecuter

Question

下面的 tcp 客户端代码基本上是循环的，直到它能够与 tcp 服务器建立连接。在 vs 代码中调试时，我注意到每次调用 create_connection() 似乎都会创建一个新的 ThreadPoolExecutor 实例。当我的 tcp 服务器未运行时，这可能会造成问题，因为似乎将创建无限数量的实例。当服务器未运行时，处理多次调用 create_connection() 的正确方法是什么？

async def do_tcp_connect(host, port, queue, shutdown_event, callbacks, connection_list):

    logger = logging.getLogger('do_tcp_connect')
    logger.debug('do_tcp_connect(): started')

    while True:
        try:
            transport, protocol = await asyncio.get_running_loop().create_connection(lambda: asyncio_callback(host, port, queue, shutdown_event, callbacks, connection_list), host, port)
            return
        except (KeyboardInterrupt, SystemExit):
            raise
        except asyncio.CancelledError as ex:
            logger.error("network_queue_consumer(): CancelledError: A asyncio coroutine task was cancelled, error={}".format(ex))
        except AttributeError as ex:
            logger.error("network_queue_consumer(): AttributeError: An attribute reference or assignment has failed, error:{}".format(ex))
        except OSError as ex:
            logger.error("network_queue_consumer(): OSError error:{}".format(ex))
        except asyncio.TimeoutError:
            continue
        except Exception as ex:
            logger.error('network_queue_consumer(): Exception occurred, error={}'.format(ex))
        except:
            logger.error("network_queue_consumer(): Unexpected error: {}".format(sys.exc_info()[0]))
            raise

        await asyncio.sleep(5.0)

    logger.debug('do_tcp_connect(): completed')

标签： python-asynciopython-3.7

Answer 1

我查看了 asyncio 源代码，发现 ThreadPoolExecutor 在下面的方法中被实例化。调用 loop.set_default_executor(executor) 似乎解决了我遇到的问题。

asyncio:base_events.py:742
    def run_in_executor(self, executor, func, *args):
        self._check_closed()
        if self._debug:
            self._check_callback(func, 'run_in_executor')
        if executor is None:
            executor = self._default_executor
            if executor is None:
                executor = concurrent.futures.ThreadPoolExecutor()
                self._default_executor = executor
        return futures.wrap_future(
            executor.submit(func, *args), loop=self)