add channel pool

sundy-li · Jul 27, 2017 · 0e6fe54 · 0e6fe54
1 parent d3fd4b8
commit 0e6fe54
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/monitor/client.go b/monitor/client.go
@@ -115,7 +115,7 @@ func (client *KafkaClient) Start() {
 	go func() {
 		defer client.wgProcessor.Done()
 		for msg := range client.messageChannel {
-			go client.RefreshConsumerOffset(msg)
+			client.RefreshConsumerOffset(msg)
 		}
 	}()
 	go func() {
@@ -154,8 +154,14 @@ func (client *KafkaClient) Start() {
 		client.wgFanIn.Add(2)
 		go func() {
 			defer client.wgFanIn.Done()
+			//20 pool per partition
+			var pool = make(chan struct{}, 20)
 			for msg := range pconsumer.Messages() {
-				client.messageChannel <- msg
+				pool <- struct{}{}
+				go func() {
+					client.messageChannel <- msg
+					<-pool
+				}()
 			}
 		}()
 		go func() {