1
使用GOCQL(Golang,Cassandra),我完成128個請求,然後一切都掛起。我相信getTicksForCassandraKey()函數我正確地發佈查詢,但不知道。 GOCQL只支持多達128個併發查詢,所以我一定在做錯事。所有查詢都是讀取。gocql阻止太多的併發讀取請求(golang,Cassandra)
主要代碼:
inboundChannel := make(chan []bson.M, 30)
maxGoRoutinesCount := 30
chunkSize := int(math.Floor(float64(len(cassandraKeys))/float64(maxGoRoutinesCount)))
log.Println("Chunk size will be: ", chunkSize)
log.Println("Cassandra Keys length: ", cassandraKeys)
idx := 0
for idx < len(cassandraKeys) {
log.Println("idx: ", idx)
chunkOfKeys := cassandraKeys[idx:(idx + chunkSize)]
idx += chunkSize
go func(keys []string) {
log.Println("Received analysisKey on outbound channel. About to query this many keys: ", len(keys))
//Cassandra session can handle up to 128 concurrent queries
for _, c := range keys {
processedTicks, err := getTicksForCassandraKey(*session, c, startTime)
if err != nil {
log.Println("Error returning.")
return
}
log.Println("Finished query. About to post to inboundChannel for key: ", c)
inboundChannel <- processedTicks
}
}(chunkOfKeys)
}
processedIndex := 0
for processedTicks := range inboundChannel {
ticks = append(ticks, processedTicks...)
log.Println("Got processed ticks from inboundChannel: ", processedIndex)
processedIndex += 1
}
log.Println("End of function.")
代碼以getTicksForCassandraKey是:
func getTicksForCassandraKey(cassandraSession gocql.Session, cassandraAnalysisKey string, startTime time.Time) (ticks []bson.M, err error) {
log.Println("getTicksForCassandraKey: ", cassandraAnalysisKey)
cassandraQuery := "select * from analysisdata where analysis_key = '" + cassandraAnalysisKey + "' and time > ? ALLOW FILTERING;"
q := cassandraSession.Query(cassandraQuery, startTime)
iter := q.Iter()
var rawData string
var rawAnalysisKey string
var rawTime time.Time
for iter.Scan(&rawAnalysisKey, &rawTime, &rawData) {
processedAlgoTick, processingErr := processAlgoTick(rawAnalysisKey, rawTime, rawData)
if processingErr != nil {
err = processingErr
return
}
ticks = append(ticks, processedAlgoTick)
}
iterCloseErr := iter.Close()
q.Release()
log.Println("Closed iter for analysis key: ", cassandraAnalysisKey)
if iterCloseErr != nil {
log.Println("Cassandra Iterator.Close() Error:", iterCloseErr)
}
return
}