Эффективное чтение строк из сжатого, разбитого на фрагменты потока HTTP по мере их поступления.

Благодаря Martijn Pieters answer я перестал работать с поведением запросов python и искал совершенно другой подход.

В итоге я использовал pyCurl. Вы можете использовать его аналогично циклу select + recv, не инвертируя поток управления и не передавая управление выделенному циклу ввода-вывода, как в Tornado и т. д. Таким образом, легко использовать генератор, который выдает новые строки, как только они приходят. - без дополнительной буферизации в промежуточных слоях, которая может привести к задержке или дополнительным потокам, запускающим цикл ввода-вывода.

В то же время он достаточно высокоуровневый, поэтому вам не нужно беспокоиться о кодировании передачи по частям, шифровании SSL или сжатии gzip.

Это был мой старый код, где chunk_size=1 приводил к загрузке процессора на 45%, а chunk_size>1 приводил к дополнительной задержке.

import requests
class RequestsHTTPStream(object):
    def __init__(self, url):
        self.url = url

    def iter_lines(self):
        headers = {'Cache-Control':'no-cache',
                   'Accept': 'text/event-stream',
                   'Accept-Encoding': 'gzip'}
        response = requests.get(self.url, stream=True, headers=headers)
        return response.iter_lines(chunk_size=1)

Вот мой новый код, основанный на pyCurl: (К сожалению, стиль curl_easy_* perform полностью блокируется, что затрудняет вывод строк между ними без использования потоков. Поэтому я использую методы curl_multi_*)

import pycurl
import urllib2
import httplib
import StringIO

class CurlHTTPStream(object):
    def __init__(self, url):
        self.url = url
        self.received_buffer = StringIO.StringIO()

        self.curl = pycurl.Curl()
        self.curl.setopt(pycurl.URL, url)
        self.curl.setopt(pycurl.HTTPHEADER, ['Cache-Control: no-cache', 'Accept: text/event-stream'])
        self.curl.setopt(pycurl.ENCODING, 'gzip')
        self.curl.setopt(pycurl.CONNECTTIMEOUT, 5)
        self.curl.setopt(pycurl.WRITEFUNCTION, self.received_buffer.write)

        self.curlmulti = pycurl.CurlMulti()
        self.curlmulti.add_handle(self.curl)

        self.status_code = 0

    SELECT_TIMEOUT = 10

    def _any_data_received(self):
        return self.received_buffer.tell() != 0

    def _get_received_data(self):
        result = self.received_buffer.getvalue()
        self.received_buffer.truncate(0)
        self.received_buffer.seek(0)
        return result

    def _check_status_code(self):
        if self.status_code == 0:
            self.status_code = self.curl.getinfo(pycurl.HTTP_CODE)
        if self.status_code != 0 and self.status_code != httplib.OK:
            raise urllib2.HTTPError(self.url, self.status_code, None, None, None)

    def _perform_on_curl(self):
        while True:
            ret, num_handles = self.curlmulti.perform()
            if ret != pycurl.E_CALL_MULTI_PERFORM:
                break
        return num_handles

    def _iter_chunks(self):
        while True:
            remaining = self._perform_on_curl()
            if self._any_data_received():
                self._check_status_code()
                yield self._get_received_data()
            if remaining == 0:
                break
            self.curlmulti.select(self.SELECT_TIMEOUT)

        self._check_status_code()
        self._check_curl_errors()

    def _check_curl_errors(self):
        for f in self.curlmulti.info_read()[2]:
            raise pycurl.error(*f[1:])

    def iter_lines(self):
        chunks = self._iter_chunks()
        return self._split_lines_from_chunks(chunks)

    @staticmethod
    def _split_lines_from_chunks(chunks):
        #same behaviour as requests' Response.iter_lines(...)

        pending = None
        for chunk in chunks:

            if pending is not None:
                chunk = pending + chunk
            lines = chunk.splitlines()

            if lines and lines[-1] and chunk and lines[-1][-1] == chunk[-1]:
                pending = lines.pop()
            else:
                pending = None

            for line in lines:
                yield line

        if pending is not None:
            yield pending

Этот код пытается получить как можно больше байтов из входящего потока, без ненужной блокировки, если их немного. Для сравнения, загрузка процессора составляет около 0,2%.

Это не requests' вина, что ваши iter_lines() вызовы блокируются.

Метод Response.iter_lines() вызывает Response.iter_content(), что звонки urllib3-х HTTPResponse.stream(), который вызывает HTTPResponse.read().

Эти вызовы проходят по размеру фрагмента, который передается в сокет как self._fp.read(amt). Это проблематичный вызов, так как self._fp — это файловый объект, созданный socket.makefile() ( как это делает модуль httplib); и этот вызов .read() будет блокироваться до тех пор, пока не будет прочитано amt (сжатых) байтов.

Этот низкоуровневый файловый объект сокета поддерживает вызов .readline(), который будет работать более эффективно, но urllib3 не может использовать этот вызов при обработке сжатых данных; терминаторы строк не будут видны в сжатом потоке.

К сожалению, urllib3 не будет вызывать self._fp.readline(), если ответ также не сжат; то, как структурированы вызовы, было бы трудно передать, что вы хотите прочитать в режиме буферизации строк, а не в режиме буферизации фрагментов, как это есть.

Я должен сказать, что HTTP — не лучший протокол для потоковой передачи событий; Я бы использовал другой протокол для этого. На ум приходят веб-сокеты или собственный протокол для вашего конкретного случая использования.