add stop

dq/consumer.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/zeromicro/go-zero/core/hash"
 	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/queue"
 	"github.com/zeromicro/go-zero/core/service"
 	"github.com/zeromicro/go-zero/core/stores/redis"
 )
@@ -19,6 +20,10 @@ const (
 var maxCheckBytes = getMaxTimeLen()
 
 type (
+	ConsumeHandler interface {
+		Consume(message string) error
+	}
+
 	Consume func(body []byte)
 
 	Consumer interface {
@@ -42,6 +47,43 @@ func NewConsumer(c DqConf) Consumer {
 	}
 }
 
+func NewService(c DqConf, handler ConsumeHandler) []queue.MessageQueue {
+	var nodes []*consumerNode
+	for _, node := range c.Beanstalks {
+		nodes = append(nodes, newConsumerNode(node.Endpoint, node.Tube))
+	}
+	cluster := &consumerCluster{
+		nodes: nodes,
+		red:   c.Redis.NewRedis(),
+	}
+	guardedConsume := func(body []byte) {
+		key := hash.Md5Hex(body)
+		body, ok := cluster.unwrap(body)
+		if !ok {
+			logx.Errorf("discarded: %q", string(body))
+			return
+		}
+
+		ok, err := cluster.red.SetnxEx(key, guardValue, expiration)
+		if err != nil {
+			logx.Error(err)
+		} else if ok {
+			handler.Consume(string(body))
+		}
+	}
+
+	var services []queue.MessageQueue
+
+	for _, node := range nodes {
+		services = append(services, consumeService{
+			c:       node,
+			consume: guardedConsume,
+		})
+	}
+
+	return services
+}
+
 func (c *consumerCluster) Consume(consume Consume) {
 	guardedConsume := func(body []byte) {
 		key := hash.Md5Hex(body)

rabbitmq/sender.go

@@ -49,3 +49,8 @@ func (q *RabbitMqSender) Send(exchange string, routeKey string, msg []byte) erro
 		},
 	)
 }
+
+func (q *RabbitMqSender) Stop() {
+	q.channel.Close()
+	q.conn.Close()
+}