[ISSUE #953] fix limiter with goroutine cover (#952)

* fix limiter with goroutine cover * fix limiter with goroutine cover Co-authored-by: 鲁扬 <qingxiang.mqx@alibaba-inc.com>
apache · Nov 2, 2022 · 96f00c4 · 96f00c4
1 parent 8afd69f
commit 96f00c4
Showing 1 changed file with 4 additions and 9 deletions.
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
@@ -1046,10 +1046,8 @@ func (pc *pushConsumer) consumeMessageConcurrently(pq *processQueue, mq *primiti
 
 	limiter := pc.option.Limiter
 	limiterOn := limiter != nil
-	if !limiterOn {
-		if _, ok := pc.crCh[mq.Topic]; !ok {
-			pc.crCh[mq.Topic] = make(chan struct{}, pc.defaultConsumer.option.ConsumeGoroutineNums)
-		}
+	if _, ok := pc.crCh[mq.Topic]; !ok {
+		pc.crCh[mq.Topic] = make(chan struct{}, pc.defaultConsumer.option.ConsumeGoroutineNums)
 	}
 
 	for count := 0; count < len(msgs); count++ {
@@ -1065,9 +1063,8 @@ func (pc *pushConsumer) consumeMessageConcurrently(pq *processQueue, mq *primiti
 
 		if limiterOn {
 			limiter(utils.WithoutNamespace(mq.Topic))
-		} else {
-			pc.crCh[mq.Topic] <- struct{}{}
 		}
+		pc.crCh[mq.Topic] <- struct{}{}
 
 		go primitive.WithRecover(func() {
 			defer func() {
@@ -1077,9 +1074,7 @@ func (pc *pushConsumer) consumeMessageConcurrently(pq *processQueue, mq *primiti
 						rlog.LogKeyConsumerGroup: pc.consumerGroup,
 					})
 				}
-				if !limiterOn {
-					<-pc.crCh[mq.Topic]
-				}
+				<-pc.crCh[mq.Topic]
 			}()
 		RETRY:
 			if pq.IsDroppd() {