feat(mq): add key field

Author: huangsheng0415

example/kq/producer/produce.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -41,7 +42,12 @@ func main() {
 		}
 
 		fmt.Println(string(body))
-		if err := pusher.Push(string(body)); err != nil {
+		if err := pusher.Push(context.Background(), string(body)); err != nil {
+			log.Fatal(err)
+		}
+
+		fmt.Println(string(body))
+		if err := pusher.KPush(context.Background(), "test", string(body)); err != nil {
 			log.Fatal(err)
 		}
 	}

kq/pusher.go

@@ -86,16 +86,29 @@ func (p *Pusher) Name() string {
 	return p.topic
 }
 
+// KPush sends a message to the Kafka topic.
+func (p *Pusher) KPush(ctx context.Context, k, v string) error {
+	msg := kafka.Message{
+		Key:   []byte(k), // current timestamp
+		Value: []byte(v),
+	}
+	if p.executor != nil {
+		return p.executor.Add(msg, len(v))
+	} else {
+		return p.producer.WriteMessages(ctx, msg)
+	}
+}
+
 // Push sends a message to the Kafka topic.
-func (p *Pusher) Push(v string) error {
+func (p *Pusher) Push(ctx context.Context, v string) error {
 	msg := kafka.Message{
 		Key:   []byte(strconv.FormatInt(time.Now().UnixNano(), 10)), // current timestamp
 		Value: []byte(v),
 	}
 	if p.executor != nil {
 		return p.executor.Add(msg, len(v))
 	} else {
-		return p.producer.WriteMessages(context.Background(), msg)
+		return p.producer.WriteMessages(ctx, msg)
 	}
 }