具有并行处理的丢失HTTPS请求

Question

我使用以下两个类方法从questradeapi（http://www.questrade.com/api/documentation/rest-operations/market-calls/markets-quotes-id）请求信息。我有超过11000个股票符号，我请求的Questrade API批量100个符号。在

import  requests
from joblib import Parallel, delayed

def parallel_request(self, elem, result, url, key):
    response = requests.get(''.join((url, elem)), headers=self.headers)
    result.extend(response.json().get(key))

Parallel(n_jobs=-1, backend="threading")(
         delayed(self.parallel_request)(elem, self.symbol_ids_list, self.uri, 'symbols')\
         for elem in self.batch_result
     )

如果我用并行类发出110个以上的HTTPS请求，那么我得到的不是11000个输出，而是10500或10600个。所以我在并行处理中丢失了数据。请注意，我在这里使用了两个python模块，即joblib（https://github.com/joblib/joblib/issues/651）和requests（https://github.com/requests/requests）。在

下面的for循环工作得很好，所以我知道我的问题出在Parallel类上。在

^{pr2}$

如何在不丢失数据的情况下提高最后一个for循环的性能？在

更新

一个self.batch_result（简化结果）的示例可以是['AAME,ABAC,ABIL,ABIO,ACERW,ACHN,ACHV,ACRX,ACST,ACTG,ADMA,ADMP,ADOM,ADXS,ADXSW,AEHR,AEMD,AETI,AEY,AEZS,AFMD,AGFSW,AGRX,AGTC,AHPAW,AHPI,AIPT,AKER,AKTX,ALIM,ALJJ,ALQA,ALSK,ALT,AMCN,AMDA,AMMA,AMRH,AMRHW,AMRN,AMRWW,AMTX,ANDAR,ANDAW,ANTH,ANY,APDN,APDNW,APOPW,APPS,APRI,APTO,APVO,APWC,AQB,AQMS,ARCI,ARCW,ARDM,AREX,ARGS,ARLZ,ARQL,ARTW,ARTX,ASFI,ASNA,ASRV,ASTC,ATACR,ATEC,ATHX,ATLC,ATOS,ATRS,AUTO,AVEO,AVGR,AVID,AVXL,AWRE,AXAS,AXON,AXSM,AYTU,AZRX,BASI,BBOX,BBRG,BCACR,BCACW,BCLI,BDSI,BHACR,BHACW,BIOC,BIOL,BIOS,BKEP,BKYI', 'BLDP,BLIN,BLNK,BLNKW,BLPH,BLRX,BMRA,BNSO,BNTC,BNTCW,BOSC,BOXL,BPTH,BRACR,BRACW,BRPAR,BRPAW,BSPM,BSQR,BUR,BURG,BVSN,BVXVW,BWEN,BYFC,CAAS,CADC,CALI,CAPR,CARV,CASI,CASM,CATB,CATS,CBAK,CBLI,CCCL,CCCR,CCIH,CDMO,CDTI,CELGZ,CERCW,CETV,CETX,CETXW,CFBK,CFMS,CFRX,CGEN,CGIX,CGNT,CHCI,CHEK,CHEKW,CHFS,CHKE,CHMA,CHNR,CIDM,CJJD,CKPT,CLDC,CLDX,CLIR,CLIRW,CLNE,CLRB,CLRBW,CLRBZ,CLSN,CLWT,CMSSR,CMSSW,CNACR,CNACW,CNET,CNIT,CNTF,CODA,CODX,COGT,CPAH,CPLP,CPRX,CPSH,CPSS,CPST,CREG,CRIS,CRME,CRNT,CSBR,CTHR,CTIB,CTIC,CTRV,CTXR,CTXRW,CUI', 'CUR,CVONW,CXDC,CXRX,CYCC,CYHHZ,CYRN,CYTR,CYTX,CYTXW,DARE,DCAR,DCIX,DELT,DEST,DFBG,DFFN,DGLY,DHXM,DLPN,DLPNW,DMPI,DOGZ,DOTAR,DOTAW,DRAD,DRIO,DRIOW,DRRX,DRYS,DSKEW,DSWL,DTEA,DTRM,DXLG,DXYN,DYNT,DYSL,EACQW,EAGLW,EARS,EASTW,EBIO,EDAP,EFOI,EGLT,EKSO,ELECW,ELGX,ELON,ELSE,ELTK,EMITF,EMMS,ENG,ENPH,ENT,EPIX,ESEA,ESES,ESTRW,EVEP,EVGN,EVK,EVLV,EVOK,EXFO,EXXI,EYEG,EYEGW,EYES,EYESW,FCEL,FCRE,FCSC,FFHL,FLGT,FLL,FMCIR,FMCIW,FNJN,FNTEW,FORD,FORK,FPAY,FRAN,FRED,FRSX,FSACW,FSNN,FTD,FTEK,FTFT,FUV,FVE,FWP,GALT,GASS,GCVRZ,GEC']

并且self.uri只是{}，如上面的questradeapi链接所示。在

更新2

马拉的回答是一个很好的尝试，但没有给我一个更好的结果。第一次测试的结果是31356（或者10452，如果我把结果除以3），而不是10900。第二个测试只给了我0或者进程块。在

我发现Maximum allowed requests per second是20。链接：http://www.questrade.com/api/documentation/rate-limiting。在考虑新信息时，如何在不丢失数据的情况下提高最后一个for循环的性能？在

Answer 1

如果您不习惯使用joblib，那么可以尝试一些标准的库并行处理模块。在python2/3中，^{}是可用的，它提供了跨并行线程映射任务的函数。简化版本如下所示：

from multiprocessing import Pool
import requests

HEADERS = {} # define headers here

def parallel_request(symbols):
    response = requests.get('https://api01.iq.questrade.com/v1/symbols?names={}'.format(symbols), headers=HEADERS)
    return response.json()

if __name__ == '__main__':
    p = Pool()
    batch_result = ['AAME,ABAC,ABIL,...',
                    'BLDP,BLIN,BLNK,...',
                    'CUR,CVONW,CXDC,...', 
                     ...]

    p.map(parallel_request, batch_result) # will return a list of len(batch_result) responses

对于更大的任务，您可能需要使用map的异步和可编辑版本，当然，您可以向您的parallel_requests任务添加参数，以避免像我那样进行硬编码。使用Pool的一个警告是，传递给它的任何参数都必须是可选择的。在

在python3中，concurrent.futures模块在文档中有一个很好的多线程url检索示例。只要稍加努力，您就可以将该示例中的load_url替换为parallel_request函数。还有一个版本的concurrent.futures作为^{}模块后移植到python2。在

这些可能需要更多的重构工作，所以如果有一个解决方案坚持joblib，请随意选择。如果您的问题是joblib中的一个bug，那么有很多方法可以用标准库（尽管添加了一些样板文件）以多线程的方式来实现这一点。在

Answer 2

很有可能是因为一些HTTP调用由于网络负载而失败。要进行测试，请更改parallel_request：

def parallel_request(self, elem, result, url, key):
    for i in range(3):  # 3 retries
        try:
            response = requests.get(''.join((url, elem)), headers=self.headers)
        except IOError: 
            continue
        result.extend(response.json().get(key))
        return

更不可能的是：list.extend不是线程安全的。如果上面的代码片段没有帮助，请尝试用锁保护extend：

^{pr2}$