pythonマルチプロセスでは、queue.putはサブプロセスをブロックします

Question

qを10個だけ入れると、出力は以下のようになります。

if cnt>10: # kk's size is about 2000. 

Begin CountChangePercentMT1 at 2016-06-16 15:19:30.706000: 
end-in 0 2016-06-16 15:19:33.291000 
end-in 1 2016-06-16 15:19:34.990000 
end-in 2 2016-06-16 15:19:36.921000 
end 0 
end 1 
end 2 
end-in 3 2016-06-16 15:19:38.748000 
end 3 
End CountChangePercentMT1 at 2016-06-16 15:20:06.665000: 

Qに20件のデータを入れた場合、出力は以下のお気に入り：

if cnt>20: # kk's size is about 2000. 

Begin CountChangePercentMT1 at 2016-06-16 15:35:36.661000: 
end-in 0 2016-06-16 15:35:39.330000 
end-in 1 2016-06-16 15:35:40.954000 
end-in 2 2016-06-16 15:35:42.828000 
end 0 
end 1 
end 2 
end-in 3 2016-06-16 15:35:44.669000 

Qに50件のデータを入れた場合、出力は以下のお気に入り：

if cnt>50: # kk's size is about 2000. 

Begin CountChangePercentMT1 at 2016-06-16 15:36:54.518000: 
end-in 0 2016-06-16 15:36:57.583000 
end-in 1 2016-06-16 15:36:58.886000 
end-in 2 2016-06-16 15:37:00.757000 
end-in 3 2016-06-16 15:37:02.648000 

Answer 1

def do_somthing(q_out,q,i,trueStartDate,trueEndDate,preStartDate,allExistTables,stock_gainian): 
#Init conn 
conn = MySQLdb.connect(host='localhost',db='tushare',user='root',passwd='',port=3306,charset='UTF8') 
#Init df_Cc 
df_Cc = pd.DataFrame() 
df_Cc['ticker'] = pd.Series(dtype=numpy.int64,index=df_Cc.index) 
df_Cc['secShortName'] = pd.Series(dtype=numpy.object,index=df_Cc.index) 
df_Cc['percent'] = pd.Series(dtype=numpy.float64,index=df_Cc.index) 
df_Cc['startPrice'] = pd.Series(dtype=numpy.float64,index=df_Cc.index) 
df_Cc['endPrice'] = pd.Series(dtype=numpy.float64,index=df_Cc.index) 
df_Cc['startDate'] = pd.Series(dtype=numpy.object,index=df_Cc.index) 
df_Cc['endDate'] = pd.Series(dtype=numpy.object,index=df_Cc.index) 
df_Cc['gainian'] = pd.Series(dtype=numpy.object,index=df_Cc.index) 
df_Cc['shizhi'] = pd.Series(dtype=numpy.float64,index=df_Cc.index) 
df_Cc['liutong'] = pd.Series(dtype=numpy.float64,index=df_Cc.index) 
while True: 
    aa = q.get() 
    if aa != None: 
     ticker = aa[0] 
     secShortName = aa[1] 
     totalShares = aa[2] 
     nonrestfloatA = aa[3] 
     listDate = datetime.datetime.strptime(aa[4], "%Y-%m-%d") 
     stock_mktequd_name = '%s%06d' % (init.g_mktequd,ticker) 
     if (stock_mktequd_name in allExistTables) & (trueEndDate>=listDate>=trueStartDate)==False: 
      str = 'select * from %s where tradeDate>=\'%s\' and tradeDate<=\'%s\' order by tradeDate' % (stock_mktequd_name,trueStartDate.strftime("%Y-%m-%d"),trueEndDate.strftime("%Y-%m-%d")) 
      df = pd.read_sql(str,conn) 

      l = len(df_Cc) 
      df_Cc.at[l,'ticker'] = ticker 
      df_Cc.at[l,'secShortName'] = secShortName 
      nn = df.shape[0] 
      if nn>0: 
       # 
       startPrice = df.iloc[0]['preClosePrice']*GetQfq(ticker,preStartDate,conn) # if nn>=n else df.iloc[0]['preClosePrice']*1 
       endPrice = df.iloc[nn-1]['closePrice']*df.iloc[nn-1]['accumAdjFactor'] 
       percent = (endPrice-startPrice)/startPrice*100 # if startPrice!=0 else -100 
       df_Cc.at[l,'startDate'] = df.iloc[0]['tradeDate'] 
       df_Cc.at[l,'endDate'] = df.iloc[nn-1]['tradeDate'] 
       df_Cc.at[l,'percent'] = percent 
       df_Cc.at[l,'startPrice'] = startPrice 
       df_Cc.at[l,'endPrice'] = endPrice 
       df_gn = stock_gainian.query('ticker==%s' % ticker) 
       if df_gn.shape[0]==1: 
        df_Cc.at[l,'gainian'] = df_gn.iloc[0]['ref_gainian'] 
       df_Cc.at[l,'shizhi'] = totalShares*startPrice/100000000 
       df_Cc.at[l,'liutong'] = nonrestfloatA*startPrice/100000000 
    else: 
     break 

q_out.put(df_Cc) 
conn.close() 
print 'end-in %s %s' % (i,datetime.datetime.now())