go 1.20

client/rabbitmq/admin.go

@@ -11,7 +11,7 @@ type Admin struct {
 }
 
 func MustNewAdmin(c *RabbitConf) *Admin {
-	conn, err := amqp.Dial(c.Address)
+	conn, err := amqp.Dial(c.URL())
 	if err != nil {
 		log.Fatalf("failed to connect rabbitmq, error: %v", err)
 	}

client/rabbitmq/producer.go

@@ -7,14 +7,19 @@ import (
 	"time"
 
 	"github.com/go-bamboo/pkg/log"
-	"github.com/go-bamboo/pkg/queue"
 	"github.com/go-bamboo/pkg/rescue"
 	"github.com/go-kratos/kratos/v2/errors"
 	"github.com/streadway/amqp"
 	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/trace"
 )
 
+type Sender interface {
+	Name() string
+	Send(ctx context.Context, header map[string]interface{}, exchange string, routeKey string, msg []byte) error
+	Close() error
+}
+
 type (
 	RabbitMqSender struct {
 		c           *ProducerConf
@@ -36,7 +41,7 @@ type (
 	}
 )
 
-func MustNewSender(c *ProducerConf) queue.Sender {
+func MustNewSender(c *ProducerConf) Sender {
 	ctx, cf := context.WithCancel(context.TODO())
 	sender := &RabbitMqSender{
 		c:               c,
@@ -112,7 +117,7 @@ func (q *RabbitMqSender) Close() error {
 }
 
 func (q *RabbitMqSender) connect() error {
-	conn, err := amqp.Dial(q.c.Rabbit.Address)
+	conn, err := amqp.Dial(q.c.Rabbit.URL())
 	if err != nil {
 		return err
 	}

client/rocketmq/consumer.go

@@ -54,12 +54,14 @@ func MustNewQueue(c *Conf, handler ConsumeHandler) queue.MessageQueue {
 
 func NewQueue(c *Conf, handler ConsumeHandler) (queue.MessageQueue, error) {
 	q := rocketQueues{}
-	cc, err := newKafkaQueue(c, handler)
-	if err != nil {
-		log.Error(err)
-		return nil, err
+	for i := 0; i < int(c.Conns); i++ {
+		cc, err := newKafkaQueue(c, handler)
+		if err != nil {
+			log.Error(err)
+			return nil, err
+		}
+		q.queues = append(q.queues, cc)
 	}
-	q.queues = append(q.queues, cc)
 	return &q, nil
 }
 
@@ -68,15 +70,15 @@ func (q rocketQueues) Name() string {
 }
 
 func (q rocketQueues) Start(ctx context.Context) error {
-	for _, queue := range q.queues {
-		queue.Start(ctx)
+	for _, qq := range q.queues {
+		qq.Start(ctx)
 	}
 	return nil
 }
 
 func (q rocketQueues) Stop(ctx context.Context) error {
-	for _, queue := range q.queues {
-		queue.Stop(ctx)
+	for _, qq := range q.queues {
+		qq.Stop(ctx)
 	}
 	return nil
 }

go.mod

@@ -1,6 +1,6 @@
 module github.com/go-bamboo/pkg
 
-go 1.19
+go 1.20
 
 require (
 	cloud.google.com/go/storage v1.30.1