[ISSUE #1051 ] Close the channel before sending data if the consumer has been shutdown.

consumer/consumer.go

@@ -297,10 +297,6 @@ 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

@@ -125,7 +125,8 @@ func (pq *processQueue) putMessage(messages ...*primitive.MessageExt) {
 		select {
 		case <-pq.closeChan:
 			return
-		case pq.msgCh <- messages:
+		default:
+			pq.msgCh <- messages
 		}
 	}
 
@@ -293,8 +294,8 @@ func (pq *processQueue) getMessages() []*primitive.MessageExt {
 	select {
 	case <-pq.closeChan:
 		return nil
-	case mq := <-pq.msgCh:
-		return mq
+	default:
+		return <-pq.msgCh
 	}
 }
 

consumer/push_consumer.go

@@ -643,6 +643,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 	NEXT:
 		select {
 		case <-pc.done:
+			close(pq.msgCh)
 			rlog.Info("push consumer close message handle.", map[string]interface{}{
 				rlog.LogKeyConsumerGroup: pc.consumerGroup,
 			})