Merge branch 'master' into master

Author: kevwan

dq/consumer.go

@@ -12,7 +12,6 @@ import (
 
 const (
 	expiration = 3600 // seconds
-	guardValue = "1"
 	tolerance  = time.Minute * 30
 )
 
@@ -45,17 +44,19 @@ func NewConsumer(c DqConf) Consumer {
 func (c *consumerCluster) Consume(consume Consume) {
 	guardedConsume := func(body []byte) {
 		key := hash.Md5Hex(body)
-		body, ok := c.unwrap(body)
+		taskBody, ok := c.unwrap(body)
 		if !ok {
 			logx.Errorf("discarded: %q", string(body))
 			return
 		}
 
-		ok, err := c.red.SetnxEx(key, guardValue, expiration)
+		redisLock := redis.NewRedisLock(c.red, key)
+		redisLock.SetExpire(expiration)
+		ok, err := redisLock.Acquire()
 		if err != nil {
 			logx.Error(err)
 		} else if ok {
-			consume(body)
+			consume(taskBody)
 		}
 	}
 

dq/consumernode.go

@@ -1,6 +1,7 @@
 package dq
 
 import (
+	"errors"
 	"time"
 
 	"github.com/beanstalkd/go-beanstalk"
@@ -59,14 +60,25 @@ func (c *consumerNode) consumeEvents(consume Consume) {
 		}
 
 		// the error can only be beanstalk.NameError or beanstalk.ConnError
-		switch cerr := err.(type) {
-		case beanstalk.ConnError:
-			switch cerr.Err {
-			case beanstalk.ErrTimeout:
+		var cerr beanstalk.ConnError
+		switch {
+		case errors.As(err, &cerr):
+			switch {
+			case errors.Is(cerr.Err, beanstalk.ErrTimeout):
 				// timeout error on timeout, just continue the loop
-			case beanstalk.ErrBadChar, beanstalk.ErrBadFormat, beanstalk.ErrBuried, beanstalk.ErrDeadline,
-				beanstalk.ErrDraining, beanstalk.ErrEmpty, beanstalk.ErrInternal, beanstalk.ErrJobTooBig,
-				beanstalk.ErrNoCRLF, beanstalk.ErrNotFound, beanstalk.ErrNotIgnored, beanstalk.ErrTooLong:
+			case
+				errors.Is(cerr.Err, beanstalk.ErrBadChar),
+				errors.Is(cerr.Err, beanstalk.ErrBadFormat),
+				errors.Is(cerr.Err, beanstalk.ErrBuried),
+				errors.Is(cerr.Err, beanstalk.ErrDeadline),
+				errors.Is(cerr.Err, beanstalk.ErrDraining),
+				errors.Is(cerr.Err, beanstalk.ErrEmpty),
+				errors.Is(cerr.Err, beanstalk.ErrInternal),
+				errors.Is(cerr.Err, beanstalk.ErrJobTooBig),
+				errors.Is(cerr.Err, beanstalk.ErrNoCRLF),
+				errors.Is(cerr.Err, beanstalk.ErrNotFound),
+				errors.Is(cerr.Err, beanstalk.ErrNotIgnored),
+				errors.Is(cerr.Err, beanstalk.ErrTooLong):
 				// won't reset
 				logx.Error(err)
 			default:

dq/producer.go

@@ -1,10 +1,8 @@
 package dq
 
 import (
-	"bytes"
 	"log"
 	"math/rand"
-	"strconv"
 	"strings"
 	"time"
 
@@ -25,15 +23,21 @@ type (
 		Close() error
 		Delay(body []byte, delay time.Duration) (string, error)
 		Revoke(ids string) error
+
+		at(body []byte, at time.Time) (string, error)
+		delay(body []byte, delay time.Duration) (string, error)
 	}
 
 	producerCluster struct {
 		nodes []Producer
 	}
 )
 
+var rng *rand.Rand
+
 func init() {
-	rand.Seed(time.Now().UnixNano())
+	source := rand.NewSource(time.Now().UnixNano())
+	rng = rand.New(source)
 }
 
 func NewProducer(beanstalks []Beanstalk) Producer {
@@ -56,10 +60,8 @@ func NewProducer(beanstalks []Beanstalk) Producer {
 }
 
 func (p *producerCluster) At(body []byte, at time.Time) (string, error) {
-	wrapped := p.wrap(body, at)
-	return p.insert(func(node Producer) (string, error) {
-		return node.At(wrapped, at)
-	})
+	wrapped := wrap(body, at)
+	return p.at(wrapped, at)
 }
 
 func (p *producerCluster) Close() error {
@@ -73,10 +75,8 @@ func (p *producerCluster) Close() error {
 }
 
 func (p *producerCluster) Delay(body []byte, delay time.Duration) (string, error) {
-	wrapped := p.wrap(body, time.Now().Add(delay))
-	return p.insert(func(node Producer) (string, error) {
-		return node.Delay(wrapped, delay)
-	})
+	wrapped := wrap(body, time.Now().Add(delay))
+	return p.delay(wrapped, delay)
 }
 
 func (p *producerCluster) Revoke(ids string) error {
@@ -98,17 +98,29 @@ func (p *producerCluster) Revoke(ids string) error {
 	return be.Err()
 }
 
+func (p *producerCluster) at(body []byte, at time.Time) (string, error) {
+	return p.insert(func(node Producer) (string, error) {
+		return node.at(body, at)
+	})
+}
+
 func (p *producerCluster) cloneNodes() []Producer {
 	return append([]Producer(nil), p.nodes...)
 }
 
+func (p *producerCluster) delay(body []byte, delay time.Duration) (string, error) {
+	return p.insert(func(node Producer) (string, error) {
+		return node.delay(body, delay)
+	})
+}
+
 func (p *producerCluster) getWriteNodes() []Producer {
 	if len(p.nodes) <= replicaNodes {
 		return p.nodes
 	}
 
 	nodes := p.cloneNodes()
-	rand.Shuffle(len(nodes), func(i, j int) {
+	rng.Shuffle(len(nodes), func(i, j int) {
 		nodes[i], nodes[j] = nodes[j], nodes[i]
 	})
 	return nodes[:replicaNodes]
@@ -156,11 +168,3 @@ func (p *producerCluster) insert(fn func(node Producer) (string, error)) (string
 
 	return "", be.Err()
 }
-
-func (p *producerCluster) wrap(body []byte, at time.Time) []byte {
-	var builder bytes.Buffer
-	builder.WriteString(strconv.FormatInt(at.UnixNano(), 10))
-	builder.WriteByte(timeSep)
-	builder.Write(body)
-	return builder.Bytes()
-}

dq/producernode.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/beanstalkd/go-beanstalk"
+	"github.com/zeromicro/go-zero/core/logx"
 )
 
 var ErrTimeBeforeNow = errors.New("can't schedule task to past time")
@@ -27,46 +28,15 @@ func NewProducerNode(endpoint, tube string) Producer {
 }
 
 func (p *producerNode) At(body []byte, at time.Time) (string, error) {
-	now := time.Now()
-	if at.Before(now) {
-		return "", ErrTimeBeforeNow
-	}
-
-	duration := at.Sub(now)
-	return p.Delay(body, duration)
+	return p.at(wrap(body, at), at)
 }
 
 func (p *producerNode) Close() error {
 	return p.conn.Close()
 }
 
 func (p *producerNode) Delay(body []byte, delay time.Duration) (string, error) {
-	conn, err := p.conn.get()
-	if err != nil {
-		return "", err
-	}
-
-	id, err := conn.Put(body, PriNormal, delay, defaultTimeToRun)
-	if err == nil {
-		return fmt.Sprintf("%s/%s/%d", p.endpoint, p.tube, id), nil
-	}
-
-	// the error can only be beanstalk.NameError or beanstalk.ConnError
-	// just return when the error is beanstalk.NameError, don't reset
-	switch cerr := err.(type) {
-	case beanstalk.ConnError:
-		switch cerr.Err {
-		case beanstalk.ErrBadChar, beanstalk.ErrBadFormat, beanstalk.ErrBuried, beanstalk.ErrDeadline,
-			beanstalk.ErrDraining, beanstalk.ErrEmpty, beanstalk.ErrInternal, beanstalk.ErrJobTooBig,
-			beanstalk.ErrNoCRLF, beanstalk.ErrNotFound, beanstalk.ErrNotIgnored, beanstalk.ErrTooLong:
-			// won't reset
-		default:
-			// beanstalk.ErrOOM, beanstalk.ErrTimeout, beanstalk.ErrUnknown and other errors
-			p.conn.reset()
-		}
-	}
-
-	return "", err
+	return p.delay(wrap(body, time.Now().Add(delay)), delay)
 }
 
 func (p *producerNode) Revoke(jointId string) error {
@@ -96,3 +66,55 @@ func (p *producerNode) Revoke(jointId string) error {
 	// if not in this beanstalk, ignore
 	return nil
 }
+
+func (p *producerNode) at(body []byte, at time.Time) (string, error) {
+	now := time.Now()
+	if at.Before(now) {
+		return "", ErrTimeBeforeNow
+	}
+
+	duration := at.Sub(now)
+	return p.delay(body, duration)
+}
+
+func (p *producerNode) delay(body []byte, delay time.Duration) (string, error) {
+	conn, err := p.conn.get()
+	if err != nil {
+		return "", err
+	}
+
+	id, err := conn.Put(body, PriNormal, delay, defaultTimeToRun)
+	if err == nil {
+		return fmt.Sprintf("%s/%s/%d", p.endpoint, p.tube, id), nil
+	}
+
+	// the error can only be beanstalk.NameError or beanstalk.ConnError
+	// just return when the error is beanstalk.NameError, don't reset
+	var cerr beanstalk.ConnError
+	switch {
+	case errors.As(err, &cerr):
+		switch {
+		case
+			errors.Is(cerr.Err, beanstalk.ErrBadChar),
+			errors.Is(cerr.Err, beanstalk.ErrBadFormat),
+			errors.Is(cerr.Err, beanstalk.ErrBuried),
+			errors.Is(cerr.Err, beanstalk.ErrDeadline),
+			errors.Is(cerr.Err, beanstalk.ErrDraining),
+			errors.Is(cerr.Err, beanstalk.ErrEmpty),
+			errors.Is(cerr.Err, beanstalk.ErrInternal),
+			errors.Is(cerr.Err, beanstalk.ErrJobTooBig),
+			errors.Is(cerr.Err, beanstalk.ErrNoCRLF),
+			errors.Is(cerr.Err, beanstalk.ErrNotFound),
+			errors.Is(cerr.Err, beanstalk.ErrNotIgnored),
+			errors.Is(cerr.Err, beanstalk.ErrTooLong):
+			// won't reset
+		default:
+			// beanstalk.ErrOOM, beanstalk.ErrTimeout, beanstalk.ErrUnknown and other errors
+			p.conn.reset()
+		}
+	default:
+		logx.Error(err)
+	}
+
+	return "", err
+}

dq/wrapper.go

@@ -0,0 +1,15 @@
+package dq
+
+import (
+	"bytes"
+	"strconv"
+	"time"
+)
+
+func wrap(body []byte, at time.Time) []byte {
+	var builder bytes.Buffer
+	builder.WriteString(strconv.FormatInt(at.UnixNano(), 10))
+	builder.WriteByte(timeSep)
+	builder.Write(body)
+	return builder.Bytes()
+}

example/dq/producer/producer.go example/dq/producer/cluster/cluster.go

@@ -0,0 +1,20 @@
+package main
+
+import (
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/zeromicro/go-queue/dq"
+)
+
+func main() {
+	producer := dq.NewProducerNode("localhost:11300", "tube")
+
+	for i := 1000; i < 1005; i++ {
+		_, err := producer.Delay([]byte(strconv.Itoa(i)), time.Second*5)
+		if err != nil {
+			fmt.Println(err)
+		}
+	}
+}

example/dq/producer/node/producer.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/zeromicro/go-queue/kq"
@@ -11,7 +12,7 @@ func main() {
 	var c kq.KqConf
 	conf.MustLoad("config.yaml", &c)
 
-	q := kq.MustNewQueue(c, kq.WithHandle(func(k, v string) error {
+	q := kq.MustNewQueue(c, kq.WithHandle(func(ctx context.Context, k, v string) error {
 		fmt.Printf("=> %s\n", v)
 		return nil
 	}))

example/kq/consumer/queue.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -40,8 +41,11 @@ func main() {
 			log.Fatal(err)
 		}
 
-		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)
+		}
+
+		if err := pusher.KPush(context.Background(), "test", string(body)); err != nil {
 			log.Fatal(err)
 		}
 	}

example/kq/producer/produce.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/zeromicro/go-queue/natsq"
+)
+
+type MyConsumer struct {
+	Channel string
+}
+
+func (c *MyConsumer) HandleMessage(m *natsq.Msg) error {
+	fmt.Printf("%s Received %s's a message: %s\n", c.Channel, m.Subject, string(m.Data))
+	return nil
+}
+
+func main() {
+
+	mc1 := &MyConsumer{Channel: "vipUpgrade"}
+	mc2 := &MyConsumer{Channel: "taskFinish"}
+
+	c := &natsq.NatsConfig{
+		ServerUri: "nats://127.0.0.1:4222",
+	}
+
+	//JetMode
+	// cq := []*natsq.ConsumerQueue{
+	// 	{
+	// 		Consumer:   mc1,
+	// 		QueueName:  "vipUpgrade",
+	// 		StreamName: "ccc",
+	// 		Subjects:   []string{"ddd", "eee"},
+	// 	},
+	// 	{
+	// 		Consumer:   mc2,
+	// 		QueueName:  "taskFinish",
+	// 		StreamName: "ccc",
+	// 		Subjects:   []string{"ccc", "eee"},
+	// 	},
+	// }
+	//q := natsq.MustNewConsumerManager(c, cq, natsq.NatJetMode)
+
+	//DefaultMode
+	cq := []*natsq.ConsumerQueue{
+		{
+			Consumer:  mc1,
+			QueueName: "vipUpgrade",
+			Subjects:  []string{"ddd", "eee"},
+		},
+		{
+			Consumer:  mc2,
+			QueueName: "taskFinish",
+			Subjects:  []string{"ccc", "eee"},
+		},
+	}
+	q := natsq.MustNewConsumerManager(c, cq, natsq.NatDefaultMode)
+	q.Start()
+	defer q.Stop()
+}

example/natsq/consumer/consumer.go

@@ -0,0 +1,68 @@
+package main
+
+import (
+	"log"
+	"math/rand"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/zeromicro/go-queue/natsq"
+)
+
+func main() {
+
+	c := natsq.NatsConfig{
+		ServerUri: "127.0.0.1:4222",
+	}
+
+	// Default Mode
+	p, _ := natsq.NewDefaultProducer(&c)
+	for i := 0; i < 3; i++ {
+		payload := randBody()
+		err := p.Publish(randSub(), payload)
+		if err != nil {
+			log.Fatalf("Error publishing message: %v", err)
+		} else {
+			log.Printf("Published message: %s", string(payload))
+		}
+	}
+	p.Close()
+
+	// JetMode
+	j, _ := natsq.NewJetProducer(&c)
+	j.CreateOrUpdateStream(jetstream.StreamConfig{
+		Name:     "ccc",
+		Subjects: []string{"ccc", "ddd", "eee"},
+		Storage:  jetstream.FileStorage,
+		NoAck:    false,
+	})
+	for i := 0; i < 3; i++ {
+		payload := randBody()
+		err := j.Publish(randSub(), payload)
+		if err != nil {
+			log.Fatalf("Error publishing message: %v", err)
+		} else {
+			log.Printf("Published message: %s", string(payload))
+		}
+	}
+	j.Close()
+}
+
+func randSub() string {
+	source := rand.NewSource(time.Now().UnixNano())
+    // 创建一个新的随机数生成器
+    rng := rand.New(source)
+	strings := []string{"ccc", "ddd", "eee"}
+	randomIndex := rng.Intn(len(strings))
+	return strings[randomIndex]
+}
+
+func randBody() []byte {
+	charSet := "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+	length := 10
+	result := make([]byte, length)
+	for i := range result {
+		result[i] = charSet[rand.Intn(len(charSet))]
+	}
+	return result
+}

example/natsq/publisher/publisher.go

@@ -44,12 +44,13 @@ func main() {
 }
 
 func randSub() string {
-	rand.Seed(time.Now().UnixNano())
+	source := rand.NewSource(time.Now().UnixNano())
+    rng := rand.New(source)
 	charSet := "abc"
 	length := 1
 	result := make([]byte, length)
 	for i := range result {
-		result[i] = charSet[rand.Intn(len(charSet))]
+		result[i] = charSet[rng.Intn(len(charSet))]
 	}
 	return string(result)
 }

example/stan/publisher/producer.go

@@ -1,13 +1,66 @@
 module github.com/zeromicro/go-queue
 
-go 1.16
+go 1.20
 
 require (
 	github.com/beanstalkd/go-beanstalk v0.2.0
+	github.com/nats-io/nats.go v1.34.1
+	github.com/nats-io/stan.go v0.10.4
+	github.com/rabbitmq/amqp091-go v1.10.0
+	github.com/segmentio/kafka-go v0.4.47
+	github.com/stretchr/testify v1.9.0
+	github.com/zeromicro/go-zero v1.6.6
+	go.opentelemetry.io/otel v1.19.0
+)
+
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/fatih/color v1.17.0 // indirect
+	github.com/go-logr/logr v1.3.0 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.18.0 // indirect
+	github.com/klauspost/compress v1.17.9 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
 	github.com/nats-io/nats-server/v2 v2.9.15 // indirect
 	github.com/nats-io/nats-streaming-server v0.25.3 // indirect
-	github.com/nats-io/stan.go v0.10.4
-	github.com/rabbitmq/amqp091-go v1.8.0
-	github.com/segmentio/kafka-go v0.4.38
-	github.com/zeromicro/go-zero v1.4.3
+	github.com/nats-io/nkeys v0.4.7 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/openzipkin/zipkin-go v0.4.2 // indirect
+	github.com/pelletier/go-toml/v2 v2.2.2 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_golang v1.18.0 // indirect
+	github.com/prometheus/client_model v0.5.0 // indirect
+	github.com/prometheus/common v0.45.0 // indirect
+	github.com/prometheus/procfs v0.12.0 // indirect
+	github.com/redis/go-redis/v9 v9.5.3 // indirect
+	github.com/spaolacci/murmur3 v1.1.0 // indirect
+	go.opentelemetry.io/otel/exporters/jaeger v1.17.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.19.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.19.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0 // indirect
+	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.19.0 // indirect
+	go.opentelemetry.io/otel/exporters/zipkin v1.19.0 // indirect
+	go.opentelemetry.io/otel/metric v1.19.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.19.0 // indirect
+	go.opentelemetry.io/otel/trace v1.19.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
+	go.uber.org/automaxprocs v1.5.3 // indirect
+	golang.org/x/crypto v0.24.0 // indirect
+	golang.org/x/net v0.26.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
+	golang.org/x/text v0.16.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
+	google.golang.org/grpc v1.64.0 // indirect
+	google.golang.org/protobuf v1.34.2 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.mod

@@ -9,16 +9,18 @@ const (
 
 type KqConf struct {
 	service.ServiceConf
-	Brokers     []string
-	Group       string
-	Topic       string
-	Offset      string `json:",options=first|last,default=last"`
-	Conns       int    `json:",default=1"`
-	Consumers   int    `json:",default=8"`
-	Processors  int    `json:",default=8"`
-	MinBytes    int    `json:",default=10240"`    // 10K
-	MaxBytes    int    `json:",default=10485760"` // 10M
-	Username    string `json:",optional"`
-	Password    string `json:",optional"`
-	ForceCommit bool   `json:",default=true"`
+	Brokers       []string
+	Group         string
+	Topic         string
+	CaFile        string `json:",optional"`
+	Offset        string `json:",options=first|last,default=last"`
+	Conns         int    `json:",default=1"`
+	Consumers     int    `json:",default=8"`
+	Processors    int    `json:",default=8"`
+	MinBytes      int    `json:",default=10240"`    // 10K
+	MaxBytes      int    `json:",default=10485760"` // 10M
+	Username      string `json:",optional"`
+	Password      string `json:",optional"`
+	ForceCommit   bool   `json:",default=true"`
+	CommitInOrder bool   `json:",default=false"`
 }

go.sum

@@ -0,0 +1,34 @@
+package internal
+
+import "github.com/segmentio/kafka-go"
+
+type Message struct {
+	*kafka.Message
+}
+
+func NewMessage(msg *kafka.Message) *Message {
+	return &Message{Message: msg}
+}
+
+func (m *Message) GetHeader(key string) string {
+	for _, h := range m.Headers {
+		if h.Key == key {
+			return string(h.Value)
+		}
+	}
+	return ""
+}
+
+func (m *Message) SetHeader(key, val string) {
+	// Ensure uniqueness of keys
+	for i := 0; i < len(m.Headers); i++ {
+		if m.Headers[i].Key == key {
+			m.Headers = append(m.Headers[:i], m.Headers[i+1:]...)
+			i--
+		}
+	}
+	m.Headers = append(m.Headers, kafka.Header{
+		Key:   key,
+		Value: []byte(val),
+	})
+}

kq/config.go

@@ -0,0 +1,57 @@
+package internal
+
+import (
+	"testing"
+
+	"github.com/segmentio/kafka-go"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMessageGetHeader(t *testing.T) {
+	testCases := []struct {
+		name     string
+		msg      *Message
+		key      string
+		expected string
+	}{
+		{
+			name: "exists",
+			msg: &Message{
+				Message: &kafka.Message{Headers: []kafka.Header{
+					{Key: "foo", Value: []byte("bar")},
+				}}},
+			key:      "foo",
+			expected: "bar",
+		},
+		{
+			name:     "not exists",
+			msg:      &Message{Message: &kafka.Message{Headers: []kafka.Header{}}},
+			key:      "foo",
+			expected: "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			result := tc.msg.GetHeader(tc.key)
+			assert.Equal(t, tc.expected, result)
+		})
+	}
+}
+
+func TestMessageSetHeader(t *testing.T) {
+	msg := &Message{Message: &kafka.Message{Headers: []kafka.Header{
+		{Key: "foo", Value: []byte("bar")}},
+	}}
+
+	msg.SetHeader("foo", "bar2")
+	msg.SetHeader("foo2", "bar2")
+	msg.SetHeader("foo2", "bar3")
+	msg.SetHeader("foo3", "bar4")
+
+	assert.ElementsMatch(t, msg.Headers, []kafka.Header{
+		{Key: "foo", Value: []byte("bar2")},
+		{Key: "foo2", Value: []byte("bar3")},
+		{Key: "foo3", Value: []byte("bar4")},
+	})
+}

kq/internal/message.go

@@ -0,0 +1,35 @@
+package internal
+
+import "go.opentelemetry.io/otel/propagation"
+
+var _ propagation.TextMapCarrier = (*MessageCarrier)(nil)
+
+// MessageCarrier injects and extracts traces from a types.Message.
+type MessageCarrier struct {
+	msg *Message
+}
+
+// NewMessageCarrier returns a new MessageCarrier.
+func NewMessageCarrier(msg *Message) MessageCarrier {
+	return MessageCarrier{msg: msg}
+}
+
+// Get returns the value associated with the passed key.
+func (m MessageCarrier) Get(key string) string {
+	return m.msg.GetHeader(key)
+}
+
+// Set stores the key-value pair.
+func (m MessageCarrier) Set(key string, value string) {
+	m.msg.SetHeader(key, value)
+}
+
+// Keys lists the keys stored in this carrier.
+func (m MessageCarrier) Keys() []string {
+	out := make([]string, len(m.msg.Headers))
+	for i, h := range m.msg.Headers {
+		out[i] = h.Key
+	}
+
+	return out
+}

kq/internal/message_test.go

@@ -0,0 +1,93 @@
+package internal
+
+import (
+	"testing"
+
+	"github.com/segmentio/kafka-go"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMessageCarrierGet(t *testing.T) {
+	testCases := []struct {
+		name     string
+		carrier  MessageCarrier
+		key      string
+		expected string
+	}{
+		{
+			name: "exists",
+			carrier: NewMessageCarrier(&Message{&kafka.Message{Headers: []kafka.Header{
+				{Key: "foo", Value: []byte("bar")},
+			}}}),
+			key:      "foo",
+			expected: "bar",
+		},
+		{
+			name:     "not exists",
+			carrier:  NewMessageCarrier(&Message{&kafka.Message{Headers: []kafka.Header{}}}),
+			key:      "foo",
+			expected: "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			result := tc.carrier.Get(tc.key)
+			assert.Equal(t, tc.expected, result)
+		})
+	}
+}
+
+func TestMessageCarrierSet(t *testing.T) {
+	msg := Message{&kafka.Message{Headers: []kafka.Header{
+		{Key: "foo", Value: []byte("bar")},
+	}}}
+	carrier := MessageCarrier{msg: &msg}
+
+	carrier.Set("foo", "bar2")
+	carrier.Set("foo2", "bar2")
+	carrier.Set("foo2", "bar3")
+	carrier.Set("foo3", "bar4")
+
+	assert.ElementsMatch(t, carrier.msg.Headers, []kafka.Header{
+		{Key: "foo", Value: []byte("bar2")},
+		{Key: "foo2", Value: []byte("bar3")},
+		{Key: "foo3", Value: []byte("bar4")},
+	})
+}
+
+func TestMessageCarrierKeys(t *testing.T) {
+	testCases := []struct {
+		name     string
+		carrier  MessageCarrier
+		expected []string
+	}{
+		{
+			name: "one",
+			carrier: MessageCarrier{msg: &Message{&kafka.Message{Headers: []kafka.Header{
+				{Key: "foo", Value: []byte("bar")},
+			}}}},
+			expected: []string{"foo"},
+		},
+		{
+			name:     "none",
+			carrier:  MessageCarrier{msg: &Message{&kafka.Message{Headers: []kafka.Header{}}}},
+			expected: []string{},
+		},
+		{
+			name: "many",
+			carrier: MessageCarrier{msg: &Message{&kafka.Message{Headers: []kafka.Header{
+				{Key: "foo", Value: []byte("bar")},
+				{Key: "baz", Value: []byte("quux")},
+			}}}},
+			expected: []string{"foo", "baz"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			result := tc.carrier.Keys()
+			assert.Equal(t, tc.expected, result)
+		})
+	}
+}

kq/internal/trace.go

@@ -6,109 +6,158 @@ import (
 	"time"
 
 	"github.com/segmentio/kafka-go"
+	"github.com/zeromicro/go-queue/kq/internal"
 	"github.com/zeromicro/go-zero/core/executors"
 	"github.com/zeromicro/go-zero/core/logx"
+	"go.opentelemetry.io/otel"
 )
 
 type (
-	PushOption func(options *chunkOptions)
+	PushOption func(options *pushOptions)
 
 	Pusher struct {
-		produer  *kafka.Writer
+		producer *kafka.Writer
 		topic    string
 		executor *executors.ChunkExecutor
 	}
 
-	chunkOptions struct {
+	pushOptions struct {
+		// kafka.Writer options
+		allowAutoTopicCreation bool
+
+		// executors.ChunkExecutor options
 		chunkSize     int
 		flushInterval time.Duration
+
+		// syncPush is used to enable sync push
+		syncPush bool
 	}
 )
 
+// NewPusher returns a Pusher with the given Kafka addresses and topic.
 func NewPusher(addrs []string, topic string, opts ...PushOption) *Pusher {
 	producer := &kafka.Writer{
 		Addr:        kafka.TCP(addrs...),
 		Topic:       topic,
 		Balancer:    &kafka.LeastBytes{},
 		Compression: kafka.Snappy,
 	}
+
+	var options pushOptions
+	for _, opt := range opts {
+		opt(&options)
+	}
+
+	// apply kafka.Writer options
+	producer.AllowAutoTopicCreation = options.allowAutoTopicCreation
+
 	pusher := &Pusher{
-		produer: producer,
-		topic:   topic,
+		producer: producer,
+		topic:    topic,
+	}
+
+	// if syncPush is true, return the pusher directly
+	if options.syncPush {
+		return pusher
+	}
+
+	// apply ChunkExecutor options
+	var chunkOpts []executors.ChunkOption
+	if options.chunkSize > 0 {
+		chunkOpts = append(chunkOpts, executors.WithChunkBytes(options.chunkSize))
+	}
+	if options.flushInterval > 0 {
+		chunkOpts = append(chunkOpts, executors.WithFlushInterval(options.flushInterval))
 	}
+
 	pusher.executor = executors.NewChunkExecutor(func(tasks []interface{}) {
 		chunk := make([]kafka.Message, len(tasks))
 		for i := range tasks {
 			chunk[i] = tasks[i].(kafka.Message)
 		}
-		if err := pusher.produer.WriteMessages(context.Background(), chunk...); err != nil {
+		if err := pusher.producer.WriteMessages(context.Background(), chunk...); err != nil {
 			logx.Error(err)
 		}
-	}, newOptions(opts)...)
+	}, chunkOpts...)
 
 	return pusher
 }
 
+// Close closes the Pusher and releases any resources used by it.
 func (p *Pusher) Close() error {
 	if p.executor != nil {
 		p.executor.Flush()
 	}
 
-	return p.produer.Close()
+	return p.producer.Close()
 }
 
+// Name returns the name of the Kafka topic that the Pusher is sending messages to.
 func (p *Pusher) Name() string {
 	return p.topic
 }
 
-func (p *Pusher) Push(v string) error {
+// KPush sends a message to the Kafka topic.
+func (p *Pusher) KPush(ctx context.Context, k, v string) error {
 	msg := kafka.Message{
-		Key:   []byte(strconv.FormatInt(time.Now().UnixNano(), 10)),
+		Key:   []byte(k), // current timestamp
 		Value: []byte(v),
 	}
 	if p.executor != nil {
 		return p.executor.Add(msg, len(v))
 	} else {
-		return p.produer.WriteMessages(context.Background(), msg)
+		return p.producer.WriteMessages(ctx, msg)
 	}
 }
 
-func (p *Pusher) PushWithKey(k, v string) error {
+// Push sends a message to the Kafka topic.
+func (p *Pusher) Push(ctx context.Context, v string) error {
+	return p.PushWithKey(ctx, strconv.FormatInt(time.Now().UnixNano(), 10), v)
+}
+
+// PushWithKey sends a message with the given key to the Kafka topic.
+func (p *Pusher) PushWithKey(ctx context.Context, key, v string) error {
 	msg := kafka.Message{
-		Key:   []byte(k),
+		Key:   []byte(key),
 		Value: []byte(v),
 	}
+
+	// wrap message into message carrier
+	mc := internal.NewMessageCarrier(internal.NewMessage(&msg))
+	// inject trace context into message
+	otel.GetTextMapPropagator().Inject(ctx, mc)
+
 	if p.executor != nil {
 		return p.executor.Add(msg, len(v))
 	} else {
-		return p.produer.WriteMessages(context.Background(), msg)
+		return p.producer.WriteMessages(ctx, msg)
 	}
 }
 
+// WithChunkSize customizes the Pusher with the given chunk size.
 func WithChunkSize(chunkSize int) PushOption {
-	return func(options *chunkOptions) {
+	return func(options *pushOptions) {
 		options.chunkSize = chunkSize
 	}
 }
 
+// WithFlushInterval customizes the Pusher with the given flush interval.
 func WithFlushInterval(interval time.Duration) PushOption {
-	return func(options *chunkOptions) {
+	return func(options *pushOptions) {
 		options.flushInterval = interval
 	}
 }
 
-func newOptions(opts []PushOption) []executors.ChunkOption {
-	var options chunkOptions
-	for _, opt := range opts {
-		opt(&options)
+// WithAllowAutoTopicCreation allows the Pusher to create the given topic if it does not exist.
+func WithAllowAutoTopicCreation() PushOption {
+	return func(options *pushOptions) {
+		options.allowAutoTopicCreation = true
 	}
+}
 
-	var chunkOpts []executors.ChunkOption
-	if options.chunkSize > 0 {
-		chunkOpts = append(chunkOpts, executors.WithChunkBytes(options.chunkSize))
-	}
-	if options.flushInterval > 0 {
-		chunkOpts = append(chunkOpts, executors.WithFlushInterval(options.flushInterval))
+// WithSyncPush enables the Pusher to push messages synchronously.
+func WithSyncPush() PushOption {
+	return func(options *pushOptions) {
+		options.syncPush = true
 	}
-	return chunkOpts
 }

kq/internal/trace_test.go

@@ -2,21 +2,29 @@ package kq
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
 	"io"
 	"log"
+	"os"
 	"time"
 
 	"github.com/segmentio/kafka-go"
 	_ "github.com/segmentio/kafka-go/gzip"
 	_ "github.com/segmentio/kafka-go/lz4"
 	"github.com/segmentio/kafka-go/sasl/plain"
 	_ "github.com/segmentio/kafka-go/snappy"
+	"github.com/zeromicro/go-queue/kq/internal"
+	"github.com/zeromicro/go-zero/core/contextx"
+	"github.com/zeromicro/go-zero/core/logc"
 	"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/stat"
 	"github.com/zeromicro/go-zero/core/threading"
 	"github.com/zeromicro/go-zero/core/timex"
+	"go.opentelemetry.io/otel"
 )
 
 const (
@@ -26,12 +34,12 @@ const (
 )
 
 type (
-	ConsumeHandle func(key, value string) error
+	ConsumeHandle func(ctx context.Context, key, value string) error
 
-	ConsumeErrorHandler func(msg kafka.Message, err error)
+	ConsumeErrorHandler func(ctx context.Context, msg kafka.Message, err error)
 
 	ConsumeHandler interface {
-		Consume(key, value string) error
+		Consume(ctx context.Context, key, value string) error
 	}
 
 	queueOptions struct {
@@ -51,6 +59,7 @@ type (
 		channel          chan kafka.Message
 		producerRoutines *threading.RoutineGroup
 		consumerRoutines *threading.RoutineGroup
+		commitRunner     *threading.StableRunner[kafka.Message, kafka.Message]
 		metrics          *stat.Metrics
 		errorHandler     ConsumeErrorHandler
 	}
@@ -121,9 +130,26 @@ func newKafkaQueue(c KqConf, handler ConsumeHandler, options queueOptions) queue
 			},
 		}
 	}
+	if len(c.CaFile) > 0 {
+		caCert, err := os.ReadFile(c.CaFile)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		caCertPool := x509.NewCertPool()
+		ok := caCertPool.AppendCertsFromPEM(caCert)
+		if !ok {
+			log.Fatal(err)
+		}
+
+		readerConfig.Dialer.TLS = &tls.Config{
+			RootCAs:            caCertPool,
+			InsecureSkipVerify: true,
+		}
+	}
 	consumer := kafka.NewReader(readerConfig)
 
-	return &kafkaQueue{
+	q := &kafkaQueue{
 		c:                c,
 		consumer:         consumer,
 		handler:          handler,
@@ -133,25 +159,51 @@ func newKafkaQueue(c KqConf, handler ConsumeHandler, options queueOptions) queue
 		metrics:          options.metrics,
 		errorHandler:     options.errorHandler,
 	}
+	if c.CommitInOrder {
+		q.commitRunner = threading.NewStableRunner(func(msg kafka.Message) kafka.Message {
+			if err := q.consumeOne(context.Background(), string(msg.Key), string(msg.Value)); err != nil {
+				if q.errorHandler != nil {
+					q.errorHandler(context.Background(), msg, err)
+				}
+			}
+
+			return msg
+		})
+	}
+
+	return q
 }
 
 func (q *kafkaQueue) Start() {
-	q.startConsumers()
-	q.startProducers()
+	if q.c.CommitInOrder {
+		go q.commitInOrder()
 
-	q.producerRoutines.Wait()
-	close(q.channel)
-	q.consumerRoutines.Wait()
+		if err := q.consume(func(msg kafka.Message) {
+			if e := q.commitRunner.Push(msg); e != nil {
+				logx.Error(e)
+			}
+		}); err != nil {
+			logx.Error(err)
+		}
+	} else {
+		q.startConsumers()
+		q.startProducers()
+		q.producerRoutines.Wait()
+		close(q.channel)
+		q.consumerRoutines.Wait()
+    
+		logx.Infof("Consumer %s is closed", q.c.Name)
+	}
 }
 
 func (q *kafkaQueue) Stop() {
 	q.consumer.Close()
 	logx.Close()
 }
 
-func (q *kafkaQueue) consumeOne(key, val string) error {
+func (q *kafkaQueue) consumeOne(ctx context.Context, key, val string) error {
 	startTime := timex.Now()
-	err := q.handler.Consume(key, val)
+	err := q.handler.Consume(ctx, key, val)
 	q.metrics.Add(stat.Task{
 		Duration: timex.Since(startTime),
 	})
@@ -162,18 +214,25 @@ func (q *kafkaQueue) startConsumers() {
 	for i := 0; i < q.c.Processors; i++ {
 		q.consumerRoutines.Run(func() {
 			for msg := range q.channel {
-				if err := q.consumeOne(string(msg.Key), string(msg.Value)); err != nil {
+				// wrap message into message carrier
+				mc := internal.NewMessageCarrier(internal.NewMessage(&msg))
+				// extract trace context from message
+				ctx := otel.GetTextMapPropagator().Extract(context.Background(), mc)
+				// remove deadline and error control
+				ctx = contextx.ValueOnlyFrom(ctx)
+
+				if err := q.consumeOne(ctx, string(msg.Key), string(msg.Value)); err != nil {
 					if q.errorHandler != nil {
-						q.errorHandler(msg, err)
+						q.errorHandler(ctx, msg, err)
 					}
 
 					if !q.c.ForceCommit {
 						continue
 					}
 				}
 
-				if err := q.consumer.CommitMessages(context.Background(), msg); err != nil {
-					logx.Errorf("commit failed, error: %v", err)
+				if err := q.consumer.CommitMessages(ctx, msg); err != nil {
+					logc.Errorf(ctx, "commit failed, error: %v", err)
 				}
 			}
 		})
@@ -182,25 +241,50 @@ func (q *kafkaQueue) startConsumers() {
 
 func (q *kafkaQueue) startProducers() {
 	for i := 0; i < q.c.Consumers; i++ {
+		i := i
 		q.producerRoutines.Run(func() {
-			for {
-				msg, err := q.consumer.FetchMessage(context.Background())
-				// io.EOF means consumer closed
-				// io.ErrClosedPipe means committing messages on the consumer,
-				// kafka will refire the messages on uncommitted messages, ignore
-				if err == io.EOF || err == io.ErrClosedPipe {
-					return
-				}
-				if err != nil {
-					logx.Errorf("Error on reading message, %q", err.Error())
-					continue
-				}
+			if err := q.consume(func(msg kafka.Message) {
 				q.channel <- msg
+			}); err != nil {
+				logx.Infof("Consumer %s-%d is closed, error: %q", q.c.Name, i, err.Error())
+				return
 			}
 		})
 	}
 }
 
+func (q *kafkaQueue) consume(handle func(msg kafka.Message)) error {
+	for {
+		msg, err := q.consumer.FetchMessage(context.Background())
+		// io.EOF means consumer closed
+		// io.ErrClosedPipe means committing messages on the consumer,
+		// kafka will refire the messages on uncommitted messages, ignore
+		if err == io.EOF || errors.Is(err, io.ErrClosedPipe) {
+			return err
+		}
+		if err != nil {
+			logx.Errorf("Error on reading message, %q", err.Error())
+			continue
+		}
+
+		handle(msg)
+	}
+}
+
+func (q *kafkaQueue) commitInOrder() {
+	for {
+		msg, err := q.commitRunner.Get()
+		if err != nil {
+			logx.Error(err)
+			return
+		}
+
+		if err := q.consumer.CommitMessages(context.Background(), msg); err != nil {
+			logx.Errorf("commit failed, error: %v", err)
+		}
+	}
+}
+
 func (q kafkaQueues) Start() {
 	for _, each := range q.queues {
 		q.group.Add(each)
@@ -252,8 +336,8 @@ type innerConsumeHandler struct {
 	handle ConsumeHandle
 }
 
-func (ch innerConsumeHandler) Consume(k, v string) error {
-	return ch.handle(k, v)
+func (ch innerConsumeHandler) Consume(ctx context.Context, k, v string) error {
+	return ch.handle(ctx, k, v)
 }
 
 func ensureQueueOptions(c KqConf, options *queueOptions) {
@@ -270,8 +354,8 @@ func ensureQueueOptions(c KqConf, options *queueOptions) {
 		options.metrics = stat.NewMetrics(c.Name)
 	}
 	if options.errorHandler == nil {
-		options.errorHandler = func(msg kafka.Message, err error) {
-			logx.Errorf("consume: %s, error: %v", string(msg.Value), err)
+		options.errorHandler = func(ctx context.Context, msg kafka.Message, err error) {
+			logc.Errorf(ctx, "consume: %s, error: %v", string(msg.Value), err)
 		}
 	}
 }

kq/pusher.go

@@ -0,0 +1,11 @@
+package natsq
+
+import (
+	"github.com/nats-io/nats.go"
+)
+
+type NatsConfig struct {
+	ServerUri  string
+	ClientName string
+	Options    []nats.Option
+}

kq/queue.go

@@ -0,0 +1,174 @@
+package natsq
+
+import (
+	"context"
+	"errors"
+	"log"
+	"sync"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/queue"
+)
+
+const (
+	NatDefaultMode = iota
+	NatJetMode
+)
+
+type (
+	Msg struct {
+		Subject string
+		Data    []byte
+	}
+
+	ConsumeHandle func(m *Msg) error
+
+	// ConsumeHandler Consumer interface, used to define the methods required by the consumer
+	ConsumeHandler interface {
+		HandleMessage(m *Msg) error
+	}
+
+	// ConsumerQueue Consumer queue, used to maintain the relationship between a consumer queue
+	ConsumerQueue struct {
+		StreamName string                     // stream name
+		QueueName  string                     // queue name
+		Subjects   []string                   // Subscribe subject
+		Consumer   ConsumeHandler             // consumer object
+		JetOption  []jetstream.PullConsumeOpt // Jetstream configuration
+	}
+
+	// ConsumerManager Consumer manager for managing multiple consumer queues
+	ConsumerManager struct {
+		mutex    sync.RWMutex    // read-write lock
+		conn     *nats.Conn      // nats connect
+		mode     uint            // nats mode
+		queues   []ConsumerQueue // consumer queue list
+		options  []nats.Option   // Connection configuration items
+		doneChan chan struct{}   // close channel
+	}
+)
+
+// MustNewConsumerManager creates a new ConsumerManager instance.
+// It connects to NATS server, registers the provided consumer queues, and returns the ConsumerManager.
+// If any error occurs during the process, it logs the error and continues.
+func MustNewConsumerManager(cfg *NatsConfig, cq []*ConsumerQueue, mode uint) queue.MessageQueue {
+	sc, err := nats.Connect(cfg.ServerUri, cfg.Options...)
+	if err != nil {
+		logx.Errorf("failed to connect nats, error: %v", err)
+	}
+	cm := &ConsumerManager{
+		conn:     sc,
+		options:  cfg.Options,
+		mode:     mode,
+		doneChan: make(chan struct{}),
+	}
+	if len(cq) == 0 {
+		logx.Errorf("failed consumerQueue register to  nats, error: cq len is 0")
+	}
+	for _, item := range cq {
+		err = cm.registerQueue(item)
+		if err != nil {
+			logx.Errorf("failed to register nats, error: %v", err)
+		}
+	}
+
+	return cm
+}
+
+// Start starts consuming messages from all the registered consumer queues.
+// It launches a goroutine for each consumer queue to subscribe and process messages.
+// The method blocks until the doneChan is closed.
+func (cm *ConsumerManager) Start() {
+	cm.mutex.RLock()
+	defer cm.mutex.RUnlock()
+
+	if len(cm.queues) == 0 {
+		logx.Errorf("no consumer queues found")
+	}
+	for _, consumerQueue := range cm.queues {
+		go cm.subscribe(consumerQueue)
+	}
+	<-cm.doneChan
+}
+
+// Stop closes the NATS connection and stops the ConsumerManager.
+func (cm *ConsumerManager) Stop() {
+	if cm.conn != nil {
+		cm.conn.Close()
+	}
+}
+
+// registerQueue registers a new consumer queue with the ConsumerManager.
+// It validates the required fields of the ConsumerQueue and adds it to the list of queues.
+// If any required field is missing, it returns an error.
+func (cm *ConsumerManager) registerQueue(queue *ConsumerQueue) error {
+	cm.mutex.Lock()
+	defer cm.mutex.Unlock()
+
+	if cm.mode == NatJetMode && queue.StreamName == "" {
+		return errors.New("stream name is required")
+	}
+
+	if queue.QueueName == "" {
+		return errors.New("queue name is required")
+	}
+	if len(queue.Subjects) == 0 {
+		return errors.New("subject is required")
+	}
+	if queue.Consumer == nil {
+		return errors.New("consumer is required")
+	}
+
+	cm.queues = append(cm.queues, *queue)
+	return nil
+}
+
+// subscribe subscribes to the specified consumer queue and starts processing messages.
+// If the NATS mode is NatJetMode, it creates a JetStream consumer and consumes messages using the provided options.
+// If the NATS mode is NatDefaultMode, it subscribes to the specified subjects using the queue name.
+// The method blocks until the doneChan is closed.
+func (cm *ConsumerManager) subscribe(queue ConsumerQueue) {
+	ctx := context.Background()
+	if cm.mode == NatJetMode {
+		js, _ := jetstream.New(cm.conn)
+		stream, err := js.Stream(ctx, "ccc")
+		if err != nil {
+			log.Fatalf("Error creating stream: %v", err)
+			return
+		}
+		consumer, _ := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
+			Name:           queue.QueueName,
+			AckPolicy:      jetstream.AckExplicitPolicy,
+			FilterSubjects: queue.Subjects,
+		})
+		consContext, subErr := consumer.Consume(func(msg jetstream.Msg) {
+			err := queue.Consumer.HandleMessage(&Msg{Subject: msg.Subject(), Data: msg.Data()})
+			if err != nil {
+				logx.Errorf("error handling message: %v", err.Error())
+			} else {
+				msg.Ack()
+			}
+		}, queue.JetOption...)
+		if subErr != nil {
+			logx.Errorf("error subscribing to queue %s: %v", queue.QueueName, subErr.Error())
+			return
+		}
+		defer consContext.Stop()
+	}
+	if cm.mode == NatDefaultMode {
+		for _, subject := range queue.Subjects {
+			cm.conn.QueueSubscribe(subject, queue.QueueName, func(m *nats.Msg) {
+				err := queue.Consumer.HandleMessage(&Msg{Subject: m.Subject, Data: m.Data})
+				if err != nil {
+					logx.Errorf("error handling message: %v", err.Error())
+				} else {
+					m.Ack()
+				}
+			})
+		}
+	}
+
+	<-cm.doneChan
+}

natsq/config.go

@@ -0,0 +1,88 @@
+package natsq
+
+import (
+	"context"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+type DefaultProducer struct {
+	conn *nats.Conn
+}
+
+// NewDefaultProducer creates a new default NATS producer.
+// It takes a NatsConfig as input and returns a pointer to a DefaultProducer and an error.
+// It connects to the NATS server using the provided configuration.
+func NewDefaultProducer(c *NatsConfig) (*DefaultProducer, error) {
+	sc, err := nats.Connect(c.ServerUri, c.Options...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &DefaultProducer{
+		conn: sc,
+	}, nil
+}
+
+// Publish publishes a message with the specified subject and data using the default NATS producer.
+// It takes a subject string and data byte slice as input and returns an error if the publish fails.
+func (p *DefaultProducer) Publish(subject string, data []byte) error {
+	return p.conn.Publish(subject, data)
+}
+
+// Close closes the NATS connection of the default producer.
+func (p *DefaultProducer) Close() {
+	if p.conn != nil {
+		p.conn.Close()
+	}
+}
+
+type JetProducer struct {
+	conn *nats.Conn
+	js   jetstream.JetStream
+	ctx  context.Context
+}
+
+// NewJetProducer creates a new JetStream producer.
+// It takes a NatsConfig as input and returns a pointer to a JetProducer and an error.
+// It connects to the NATS server using the provided configuration and creates a new JetStream context.
+func NewJetProducer(c *NatsConfig) (*JetProducer, error) {
+	sc, err := nats.Connect(c.ServerUri, c.Options...)
+	if err != nil {
+		return nil, err
+	}
+	js, err := jetstream.New(sc)
+	if err != nil {
+		return nil, err
+	}
+	return &JetProducer{
+		conn: sc,
+		js:   js,
+	}, nil
+}
+
+// CreateOrUpdateStream creates or updates a JetStream stream with the specified configuration.
+// It takes a jetstream.StreamConfig as input and returns an error if the operation fails.
+func (j *JetProducer) CreateOrUpdateStream(config jetstream.StreamConfig) error {
+	_, err := j.js.CreateOrUpdateStream(j.ctx, config)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// Publish publishes a message with the specified subject and data using the JetStream producer.
+// It takes a subject string and data byte slice as input and returns an error if the publish fails.
+func (j *JetProducer) Publish(subject string, data []byte) error {
+	_, err := j.js.Publish(j.ctx, subject, data)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// Close closes the NATS connection of the JetStream producer.
+func (j *JetProducer) Close() {
+	j.conn.Close()
+}

natsq/consumer.go

@@ -53,7 +53,7 @@ Kafka Pub/Sub framework
 
 ### consumer example
 
-config.json
+config.yaml
 ```yaml
 Name: kq
 Brokers: