feat(mq): open telemetry support

Author: setcy

mq/producer.go

@@ -5,6 +5,7 @@ import (
 	"crypto/md5"
 	"fmt"
 	amqp "github.com/rabbitmq/amqp091-go"
+	"go.opentelemetry.io/otel"
 )
 
 type Producer struct {
@@ -86,6 +87,10 @@ func parseOptions(options []string) map[string]string {
 
 func (p *Producer) publish(ctx context.Context, key string, msg []byte, opts map[string]string) error {
 
+	carrier := RabbitMQHeaderCarrier{}
+	otel.GetTextMapPropagator().Inject(ctx, carrier)
+	headers := amqp.Table(carrier)
+
 	err := p.Channel.PublishWithContext(
 		ctx,
 		exchangeName,
@@ -99,6 +104,7 @@ func (p *Producer) publish(ctx context.Context, key string, msg []byte, opts map
 			AppId:        p.appId,
 			UserId:       opts[UserIdKey],
 			MessageId:    fmt.Sprintf("%x", md5.Sum(msg)),
+			Headers:      headers,
 		})
 
 	return err

mq/producer_test.go

@@ -17,12 +17,12 @@ func TestSendNotify(t *testing.T) {
 		SetData("keyword4", "通知内容", "")                           // 通知内容
 	notice.Receiver("12345678")
 
-	p, err := NewProducer("sdk_test", "amqpURL")
+	p, err := NewProducer("sdk_test", "amqpURL") // 创建一个MQ连接负责消息生产
 	if err != nil {
 		t.Error(err)
 	}
 
-	err = p.PublishNotice(context.Background(), notice)
+	err = p.PublishNotice(context.Background(), notice) // 发送消息
 	if err != nil {
 		t.Error(err)
 	}

mq/propagator.go

@@ -0,0 +1,27 @@
+package mq
+
+import (
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+// RabbitMQHeaderCarrier adapts http.Header to satisfy the TextMapCarrier interface.
+type RabbitMQHeaderCarrier amqp.Table
+
+// Get returns the value associated with the passed key.
+func (hc RabbitMQHeaderCarrier) Get(key string) string {
+	return hc.Get(key)
+}
+
+// Set stores the key-value pair.
+func (hc RabbitMQHeaderCarrier) Set(key string, value string) {
+	hc.Set(key, value)
+}
+
+// Keys lists the keys stored in this carrier.
+func (hc RabbitMQHeaderCarrier) Keys() []string {
+	keys := make([]string, 0, len(hc))
+	for k := range hc {
+		keys = append(keys, k)
+	}
+	return keys
+}