Fix go routine leaks when consumer close with msg channel blocked (#642)

Co-authored-by: zhangxu16 <[email protected]>

consumer/consumer.go

@@ -305,6 +305,10 @@ func (dc *defaultConsumer) shutdown() error {
 		k := key.(primitive.MessageQueue)
 		pq := value.(*processQueue)
 		pq.WithDropped(true)
+		// close msg channel using RWMutex to make sure no data was writing
+		pq.mutex.Lock()
+		close(pq.msgCh)
+		pq.mutex.Unlock()
 		mqs = append(mqs, &k)
 		return true
 	})

consumer/process_queue.go

@@ -91,6 +91,10 @@ func (pq *processQueue) putMessage(messages ...*primitive.MessageExt) {
 		return
 	}
 	pq.mutex.Lock()
+	if pq.IsDroppd() {
+		pq.mutex.Unlock()
+		return
+	}
 	if !pq.order {
 		pq.msgCh <- messages
 	}