python - ProactorEventLoop - 值错误 : loop argument must agree with Future

Question

我正在使用这个异步项目(称为 Broker - see git )，代码如下:

   proxies = asyncio.Queue()
   broker = Broker(proxies)
   tasks = asyncio.gather(broker.find(types = ['HTTPS'], strict = True, limit = 10),
                               self.storeProxies(proxies))
   loop = asyncio.get_event_loop()
   loop.run_until_complete(tasks)

其中 self.storeProxies 是一个异步函数，其中包含

while True:
    proxy = await proxies.get()
    if proxy is None: break
    self.proxies['http://{0}:{1}'.format(proxy.host, proxy.port)] = 1

但是，我不认为它与Broker部分有关(当然不确定)。

每次我运行此代码时，在相当随机的成功次数之后，它都会失败，并显示[WinError 10038]尝试对非套接字的操作。然后我尝试做一些研究，最终得到this answer 。但是当尝试使用此代码时，我收到此错误:

ValueError:循环参数必须与 Future 一致

关于如何处理这个问题有什么想法吗？

可能有用的详细信息:

操作系统:Windows 10(版本 1809)Python版本:3.7.1

基于 Mikhail Gerasimov 的回答

async def storeProxies(self, proxies):
    logger.info("IN IT!")
    while True:
        proxy = await proxies.get()
        if proxy is None: break
        self.proxies['http://{0}:{1}'.format(proxy.host, proxy.port)] = 1

async def getfetching(self, amount):
    from proxybroker import Broker

    proxies = asyncio.Queue()
    broker = Broker(proxies)
    return await asyncio.gather(
        broker.find(
            types = ['HTTP', 'HTTPS'],
            strict = True,
            limit = 10
        ),
        self.storeProxies(proxies)
    )
def fetchProxies(self, amount):
    if os.name == 'nt':
        loop = asyncio.SelectorEventLoop() # for subprocess' pipes on Windows
        asyncio.set_event_loop(loop)
    else:
        loop = asyncio.get_event_loop()
    loop.run_until_complete(self.getfetching(amount))
    logger.info("FETCHING COMPLETE!!!!!!")
    logger.info('Proxies: {}'.format(self.proxies))

其中 fetchProxies 从另一个地方以一定的时间间隔调用。第一次工作完美，但随后“失败”并出现警告:

2019-04-06 21:04:21 [py.warnings] WARNING: d:\users\me\anaconda3\lib\site-packages\proxybroker\providers.py:78: DeprecationWarning: The object should be created from async function
  headers=get_headers(), cookies=self._cookies, loop=self._loop

2019-04-06 21:04:21 [py.warnings] WARNING: d:\users\me\anaconda3\lib\site-packages\aiohttp\connector.py:730: DeprecationWarning: The object should be created from async function
  loop=loop)

2019-04-06 21:04:21 [py.warnings] WARNING: d:\users\me\anaconda3\lib\site-packages\aiohttp\cookiejar.py:55: DeprecationWarning: The object should be created from async function
  super().__init__(loop=loop)

接下来的行为看起来像是无限循环(硬卡住不输出任何内容)。值得注意的是，在全局导入 Broker 时，Gerasimov 的示例(在评论中)也发生了这种情况。终于，我开始看到隧道的曙光。

Answer 1

看一下这部分:

proxies = asyncio.Queue()

# ...

loop.run_until_complete(tasks)

asyncio.Queue() 创建时绑定(bind)到当前事件循环(默认事件循环或使用 asyncio.set_event_loop() 设置为当前事件循环)。通常只有循环对象绑定(bind)才能管理它。如果更改当前循环，则应该重新创建对象。对于许多其他 asyncio 对象也是如此。

为了确保每个对象都绑定(bind)到新的事件循环，最好在设置并运行新的事件循环后创建与 asyncio 相关的对象。它看起来像这样:

async def main():
    proxies = asyncio.Queue()
    broker = Broker(proxies)
    return await asyncio.gather(
        broker.find(
            types = ['HTTPS'], 
            strict = True, 
            limit = amount
        ),
        self.storeProxies(proxies)
    )


if os.name == 'nt':
    loop = asyncio.ProactorEventLoop() # for subprocess' pipes on Windows
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()
loop.run_until_complete(main())

有时(很少)您必须在 main() 中放置一些导入。

<小时/>

更新:

出现问题的原因是您在 fetchProxies 调用之间更改了事件循环，但 Broker 仅导入了一次(Python 缓存导入的模块)。

Reloading Broker 对我不起作用，因此我找不到比重用您设置一次的事件循环更好的方法。

替换此代码

if os.name == 'nt':
    loop = asyncio.SelectorEventLoop() # for subprocess' pipes on Windows
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()

有了这个

if os.name == 'nt':
    loop = asyncio.get_event_loop()
    if not isinstance(loop, asyncio.SelectorEventLoop):
        loop = asyncio.SelectorEventLoop() # for subprocess' pipes on Windows
        asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()

附注

顺便说一句，您一开始就不必这样做:

if os.name == 'nt':
    loop = asyncio.SelectorEventLoop()
    asyncio.set_event_loop(loop)

Windows 已默认使用 SelectorEventLoop，但它不支持管道 ( doc )。