fix: exit wrong (#34)

consumer/consumer.go

@@ -103,10 +103,6 @@ func (nc *NSQConsumer) Run(c *consumer.ConsumerEntity) error {
 
 	_ = nc.pipeline.Close()
 
-	for _, consumer := range nc.topics {
-		close(consumer.done)
-	}
-
 	nc.wg.Wait()
 	return nil
 }
@@ -166,6 +162,13 @@ func (nc *NSQConsumer) start() {
 func (nc *NSQConsumer) Stop() {
 	logp.L().Info("Stopping nsq consumer")
 
+	// need close consumer first and then close others
+	for _, consumer := range nc.topics {
+		// close(consumer.done)
+		consumer.Close()
+		close(consumer.done)
+	}
+
 	// Stop nsq consumer
 	close(nc.done)
 }
@@ -200,14 +203,15 @@ func newNsqConsumer(opts *Options, topic string, cfg *nsq.Config, etcdConfig *et
 func (c Consumer) HandleMessage(m *nsq.Message) error {
 	m.DisableAutoResponse()
 	c.msgChan <- m
+
 	return nil
 }
 
 func (c *Consumer) router() {
 	for {
 		select {
 		case <-c.done:
-			c.Close()
+			// c.Close()
 			return
 		case m := <-c.msgChan:
 			c.queue <- &Message{

libconsumer/cmd/instance/consumer.go

@@ -239,6 +239,13 @@ func (c *Consumer) InitWithSettings(settings Settings) error {
 }
 
 func (c *Consumer) launch(settings Settings, ct consumer.Creator) error {
+	// todo: pprof
+	// 优先启动pprof
+	// pprofHandler := http.NewServeMux()
+	// pprofHandler.Handle("/debug/pprof/", http.HandlerFunc(pprof.Index))
+	// server := &http.Server{Addr: ":7070", Handler: pprofHandler}
+	// go server.ListenAndServe()
+
 	defer func() {
 		_ = logp.Sync()
 	}()

libconsumer/publisher/pipeline/output.go

@@ -72,7 +72,7 @@ func (w *clientWorker) run() {
 	for {
 		select {
 		case <-w.done:
-			w.logger.Info("clientWorker#run accep done signal")
+			w.logger.Info("clientWorker#run accept done signal")
 			return
 		case m := <-w.msgChan:
 			if err := w.client.Publish(context.TODO(), m); err != nil {