python - Errno None 无法为具有大量请求的 写入请求正文

Question

跟踪器经常收到一个错误，我很长时间都无法修复和重现。

这是我的客户的样子

class Client:
    _auth = (API_USER, API_PASSWORD)

    def __init__(self):
        self.auth = aiohttp.BasicAuth(self._auth[0], self._auth[1])

    async def send(self, url, json, debug=True,):
        try:
            async with aiohttp.ClientSession(
                    connector=aiohttp.TCPConnector(verify_ssl=False),
                    raise_for_status=True,
            ) as session:
                async with session.post(
                        url=url,
                        json=json,
                        auth=self.auth,
                ) as response:
                    if debug:
                        print(f"{url} response.status", response.status, json)
                    if response.status not in [200, 201]:
                        error = False
                        result = None
                        return error, result

                    error = False
                    result = await response.json(content_type=None)

                    if 'error' in result:
                        if debug:
                            print(f'{url} error', result, json)
                        error = True
                        result = None

                    return error, result
        except (
                aiohttp.client_exceptions.ClientResponseError,
                aiohttp.client_exceptions.ClientConnectorError,
                JSONDecodeError,
                TimeoutError,
        ) as e:
            error = True
            result = None
            return error, result


Client.send(
    url="my_url",
    json=my_json
)

请告诉我问题是什么

重现:不知道怎么办

预期行为:正常发送请求

日志/回溯:

ConnectionResetError: Cannot write to closing transport
  ...
  File "aiohttp/http_writer.py", line 67, in _write
    raise ConnectionResetError('Cannot write to closing transport')


ClientOSError: [Errno None] Can not write request body for <my_url>
  ...
  File "aiohttp/streams.py", line 588, in read
    await self._waiter

Answer 1

我已经通过创建一个新 session 解决了我的问题。之前，我在类中初始化一个 Session，然后用它发出大量的 POST 请求，我偶尔会在日志文件中找到这个“[Errno None] Can not write request body xxx”。因此，我认为我们应该为每个请求创建一个新的 session ，并在得到响应后关闭它。

# before
class TestClient():
    def __init__(self):

        self.a_url = 'http://127.0.0.1:22091/test1'
        self.b_url = 'http://127.0.0.1:22092/test2'
        self.c_url = 'http://127.0.0.1:22093/test3'

        new_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(new_loop)
        self.loop = asyncio.get_event_loop()
        self.session = self.loop.run_until_complete(self.get_session())

    async def get_session(self):
        return aiohttp.ClientSession(read_timeout=2)

    def __del__(self):
        self.loop.close()

    async def async_single_call(self):
        async_time_out = 10
        try:
            task_list = [ 
            asyncio.ensure_future(async_down(self.session, self.a_url, {'msg': "hello"*100000}, async_time_out)),
            asyncio.ensure_future(async_down(self.session, self.b_url, {'msg': "hello"*100000}, async_time_out)),
            asyncio.ensure_future(async_down(self.session, self.c_url, {'msg': "hello"*100000}, async_time_out))]
        except Exception as e:
            print(e)
        finally:
            pass


# after
class TestClient():
    def __init__(self):

        self.a_url = 'http://127.0.0.1:22091/test1'
        self.b_url = 'http://127.0.0.1:22092/test2'
        self.c_url = 'http://127.0.0.1:22093/test3'

        new_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(new_loop)
        self.loop = asyncio.get_event_loop()
     

    def get_session(self):
        return aiohttp.ClientSession(read_timeout=2)

    def __del__(self):
        self.loop.close()

    async def async_single_call(self):
        async_time_out = 10
        the_new_session = self.get_session()
        try:
            task_list = [ 
            asyncio.ensure_future(async_down(the_new_session, self.a_url, {'msg': "hello"*100000}, async_time_out)),
            asyncio.ensure_future(async_down(the_new_session, self.b_url, {'msg': "hello"*100000}, async_time_out)),
            asyncio.ensure_future(async_down(the_new_session, self.c_url, {'msg': "hello"*100000}, async_time_out))]
        except Exception as e:
            print(e)
        finally:
            await the_new_session.close()

希望能给你带来一些启发:-)