nats

src/jynats/.gitignore

@@ -0,0 +1,26 @@
+# ---> Go
+# Compiled Object files, Static and Dynamic libs (Shared Objects)
+*.o
+*.a
+*.so
+
+# Folders
+_obj
+_test
+
+# Architecture specific extensions/prefixes
+*.[568vq]
+[568vq].out
+
+*.cgo1.go
+*.cgo2.c
+_cgo_defun.c
+_cgo_gotypes.go
+_cgo_export.*
+
+_testmain.go
+
+*.exe
+*.test
+*.prof
+

src/jynats/.idea/.gitignore

@@ -0,0 +1,8 @@
+# Default ignored files
+/shelf/
+/workspace.xml
+# Editor-based HTTP Client requests
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

src/jynats/.idea/encodings.xml

@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Encoding" addBOMForNewFiles="with NO BOM" />
+</project>

src/jynats/.idea/jynats.iml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="WEB_MODULE" version="4">
+  <component name="Go" enabled="true" />
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

src/jynats/.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/jynats.iml" filepath="$PROJECT_DIR$/.idea/jynats.iml" />
+    </modules>
+  </component>
+</project>

src/jynats/README.md

@@ -0,0 +1,99 @@
+# jynats
+
+nats的go工具包，支持nats和natsjs两种模式
+
+## nats 模式
+> 此模式必须有消费者监听，否则消息直接丢弃
+
+```
+import "jygit.jydev.jianyu360.cn/BP/jynats/jnats"
+
+	jn := jnats.NewJnats(addr)
+
+	//先消费，带zip压缩，用于跨网传输节省流量
+	jn.SubZip("test", func(msg *nats.Msg) {
+		log.Println(string(msg.Data))
+		//回执消息
+		msg.Respond([]byte("receive msg:" + string(msg.Data)))
+	})
+	
+	
+		//后发布，带压缩
+	go func() {
+		for i := 0; i < 100; i++ {
+			rep, err := jn.PubReqZip("test", fmt.Sprintf("你好，ping%d", i), time.Second)
+			if err != nil {
+				log.Println(err) //消息重试机制...
+			} else {
+				log.Println(string(rep.Data))
+			}
+			time.Sleep(500 * time.Millisecond)
+		}
+	}()
+
+	
+	
+```
+
+
+## natsjs 模式
+> 此模式需要指定stream,默认的stream为jydef,管理以jydef.>类的主题，其他主题自行定义stream 
+
+```
+import "jygit.jydev.jianyu360.cn/BP/jynats/jnatsjs"
+
+	jsn := jnatsjs.NewJnatsjs(addr)
+    
+    //带压缩
+	go func() {
+		for i := 0; i < 1000000; i++ {
+			jsn.PubZip("jydef.test1", fmt.Sprintf("你好，测试%d", i))
+			time.Sleep(1500 * time.Millisecond)
+		}
+	}()
+	jsn.SubZip("jydef.test1", "aa1", func(msg *nats.Msg) {
+		msg.Ack()
+		log.Println(string(msg.Data))
+	})
+
+	//不带压缩
+	go func() {
+		for i := 0; i < 1000000; i++ {
+			jsn.Pub("jydef.test2", fmt.Sprintf("hello,%d", i))
+			time.Sleep(1500 * time.Millisecond)
+		}
+	}()
+	jsn.Sub("jydef.test2", "aa2", func(msg *nats.Msg) {
+		msg.Ack()
+		log.Println(string(msg.Data))
+	})
+
+```
+
+## nets docker启动
+```
+#js模式
+docker run -it -d -v /mntdisk/rz/nats/data:/datastore -p 4223:4223 -p 8223:8223 nats:2.10.6 -p 4223 -m 8223  -js -sd /datastore
+
+
+#nats模式
+docker network create --subnet=172.17.2.0/24 nats
+docker run -d --name=nats-1 -p 5222:5222 -p 7222:7222 -p 9222:9222 -v $(pwd)/nats.conf:/etc/nats/nats-server.conf --network=nats nats -c /etc/nats/nats-server.conf 
+
+nats.conf文件
+
+port: 5222
+monitor_port: 9222
+# 集群模式，可不定义
+cluster {
+  name: "jynats"
+  port: 7222
+  authorization {
+    user: rz
+    password: pwd
+    timeout: 5
+  }
+  routes = []
+}
+
+```

src/jynats/go.mod

@@ -0,0 +1,36 @@
+module jygit.jydev.jianyu360.cn/BP/jynats
+
+go 1.19
+
+require (
+	github.com/gogf/gf/v2 v2.5.7
+	github.com/nats-io/nats.go v1.31.0
+)
+
+require (
+	github.com/BurntSushi/toml v1.2.0 // indirect
+	github.com/clbanning/mxj/v2 v2.7.0 // indirect
+	github.com/fatih/color v1.15.0 // indirect
+	github.com/fsnotify/fsnotify v1.7.0 // indirect
+	github.com/go-logr/logr v1.2.4 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
+	github.com/grokify/html-strip-tags-go v0.0.1 // indirect
+	github.com/klauspost/compress v1.17.0 // indirect
+	github.com/magiconair/properties v1.8.6 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mattn/go-runewidth v0.0.15 // indirect
+	github.com/nats-io/nkeys v0.4.5 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/olekukonko/tablewriter v0.0.5 // indirect
+	github.com/rivo/uniseg v0.4.4 // indirect
+	go.opentelemetry.io/otel v1.14.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.14.0 // indirect
+	go.opentelemetry.io/otel/trace v1.14.0 // indirect
+	golang.org/x/crypto v0.14.0 // indirect
+	golang.org/x/net v0.17.0 // indirect
+	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/text v0.13.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

src/jynats/go.sum

@@ -0,0 +1,66 @@
+github.com/BurntSushi/toml v1.2.0 h1:Rt8g24XnyGTyglgET/PRUNlrUeu9F5L+7FilkXfZgs0=
+github.com/BurntSushi/toml v1.2.0/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
+github.com/clbanning/mxj/v2 v2.7.0 h1:WA/La7UGCanFe5NpHF0Q3DNtnCsVoxbPKuyBNHWRyME=
+github.com/clbanning/mxj/v2 v2.7.0/go.mod h1:hNiWqW14h+kc+MdF9C6/YoRfjEJoR3ou6tn/Qo+ve2s=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/fatih/color v1.15.0 h1:kOqh6YHBtK8aywxGerMG2Eq3H6Qgoqeo13Bk2Mv/nBs=
+github.com/fatih/color v1.15.0/go.mod h1:0h5ZqXfHYED7Bhv2ZJamyIOUej9KtShiJESRwBDUSsw=
+github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
+github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
+github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/logr v1.2.4 h1:g01GSCwiDw2xSZfjJ2/T9M+S6pFdcNtFYsp+Y43HYDQ=
+github.com/go-logr/logr v1.2.4/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/gogf/gf/v2 v2.5.7 h1:h+JSoD6z3d2q0uGszvtahrSm4DiM2ECyNjyTwKIo8wE=
+github.com/gogf/gf/v2 v2.5.7/go.mod h1:x2XONYcI4hRQ/4gMNbWHmZrNzSEIg20s2NULbzom5k0=
+github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/grokify/html-strip-tags-go v0.0.1 h1:0fThFwLbW7P/kOiTBs03FsJSV9RM2M/Q/MOnCQxKMo0=
+github.com/grokify/html-strip-tags-go v0.0.1/go.mod h1:2Su6romC5/1VXOQMaWL2yb618ARB8iVo6/DR99A6d78=
+github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJwgrqM=
+github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/magiconair/properties v1.8.6 h1:5ibWZ6iY0NctNGWo87LalDlEZ6R41TqbbDamhfG/Qzo=
+github.com/magiconair/properties v1.8.6/go.mod h1:y3VJvCyxH9uVvJTWEGAELF3aiYNyPKd5NZ3oSwXrF60=
+github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
+github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
+github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
+github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
+github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
+github.com/mattn/go-runewidth v0.0.15 h1:UNAjwbU9l54TA3KzvqLGxwWjHmMgBUVhBiTjelZgg3U=
+github.com/mattn/go-runewidth v0.0.15/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
+github.com/nats-io/nats.go v1.31.0 h1:/WFBHEc/dOKBF6qf1TZhrdEfTmOZ5JzdJ+Y3m6Y/p7E=
+github.com/nats-io/nats.go v1.31.0/go.mod h1:di3Bm5MLsoB4Bx61CBTsxuarI36WbhAwOm8QrW39+i8=
+github.com/nats-io/nkeys v0.4.5 h1:Zdz2BUlFm4fJlierwvGK+yl20IAKUm7eV6AAZXEhkPk=
+github.com/nats-io/nkeys v0.4.5/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=
+github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
+github.com/rivo/uniseg v0.4.4 h1:8TfxU8dW6PdqD27gjM8MVNuicgxIjxpm4K7x4jp8sis=
+github.com/rivo/uniseg v0.4.4/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
+github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
+go.opentelemetry.io/otel v1.14.0 h1:/79Huy8wbf5DnIPhemGB+zEPVwnN6fuQybr/SRXa6hM=
+go.opentelemetry.io/otel v1.14.0/go.mod h1:o4buv+dJzx8rohcUeRmWUZhqupFvzWis188WlggnNeU=
+go.opentelemetry.io/otel/sdk v1.14.0 h1:PDCppFRDq8A1jL9v6KMI6dYesaq+DFcDZvjsoGvxGzY=
+go.opentelemetry.io/otel/sdk v1.14.0/go.mod h1:bwIC5TjrNG6QDCHNWvW4HLHtUQ4I+VQDsnjhvyZCALM=
+go.opentelemetry.io/otel/trace v1.14.0 h1:wp2Mmvj41tDsyAJXiWDWpfNsOiIyd38fy85pyKcFq/M=
+go.opentelemetry.io/otel/trace v1.14.0/go.mod h1:8avnQLK+CG77yNLUae4ea2JDQ6iT+gozhnZjy/rw9G8=
+golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
+golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
+golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
+golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

src/jynats/jnats/jnats.go

@@ -0,0 +1,159 @@
+package jnats
+
+import (
+	"bytes"
+	"compress/gzip"
+	log "github.com/donnie4w/go-logger/logger"
+	"io/ioutil"
+	"time"
+
+	//"github.com/gogf/gf/v2/encoding/gcompress"
+	//"github.com/gogf/gf/v2/frame/g"
+	"github.com/nats-io/nats.go"
+)
+
+/*
+没有消费者，消息会丢掉
+1、使用reply机制，应用来控制
+2、必须先有订阅，且订阅者不能断
+*/
+
+type Jnats struct {
+	Addr string //nats服务地址
+	Nc   *nats.Conn
+}
+
+func NewJnats(addr string) *Jnats {
+	js := &Jnats{
+		Addr: addr,
+	}
+	js.ReConnect()
+	return js
+}
+
+// 连接、设置、重试
+func (j *Jnats) ReConnect() bool {
+	var err error
+	opts := []nats.Option{
+		//nats.Name(c.Name), 指定clent名字
+		// nats.SetCustomDialer(n),
+		nats.MaxReconnects(86400),
+		nats.ReconnectWait(time.Second), //默认两秒
+		nats.ReconnectBufSize(83886080), //在链接繁时的消息缓冲，默认8M
+		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
+			//g.Log().Error(context.Background(), "NATS error: %v", err)
+			log.Debug("NATS error: ", err)
+		}),
+		nats.DisconnectHandler(func(c *nats.Conn) {
+			//g.Log().Info(context.Background(), "Disconnected from NATS")
+			log.Debug("Disconnected from NATS")
+		}),
+		nats.ClosedHandler(func(c *nats.Conn) {
+			//g.Log().Info(context.Background(), "NATS connection is closed")
+			log.Debug("NATS connection is closed")
+		}),
+	}
+	if j.Nc == nil || j.Nc.IsClosed() {
+		j.Nc, err = nats.Connect(j.Addr, opts...)
+		if err != nil {
+			//g.Log().Error(context.Background(), "NATS connect error: %v", err)
+			log.Debug("NATS connect error: ", err)
+			time.Sleep(time.Second)
+			return j.ReConnect()
+		} else {
+			return true
+		}
+	} else if j.Nc.IsConnected() { //连接状态
+		return true
+	} else { //异常状态
+		j.Nc.Flush()
+		j.Nc.Drain()
+		j.Nc.Close()
+		time.Sleep(time.Second)
+		return j.ReConnect()
+	}
+	return false
+}
+
+// // 生产消息
+//
+//	func (j *Jnats) Pub(sub string, msg any) error {
+//		return j.Nc.Publish(sub, g.NewVar(msg).Bytes())
+//	}
+//
+// // 生产压缩消息
+//
+//	func (j *Jnats) PubZip(sub string, msg any) error {
+//		res, err := gcompress.Zlib(g.NewVar(msg).Bytes())
+//		if err != nil {
+//			return err
+//		} else {
+//			return j.Nc.Publish(sub, res)
+//		}
+//	}
+//
+// // 直接消费消息
+//
+//	func (j *Jnats) Sub(sub string, handle func(msg *nats.Msg)) {
+//		_, err := j.Nc.Subscribe(sub, handle)
+//		if err != nil {
+//			g.Log().Error(context.Background(), err)
+//		}
+//	}
+//
+// 直接消费压缩消息
+func (j *Jnats) SubZip(sub string, handle func(msg *nats.Msg)) {
+	_, err := j.Nc.Subscribe(sub, func(msg *nats.Msg) {
+		v := msg.Data
+		if len(v) > 0 {
+			reader, err := gzip.NewReader(bytes.NewReader(v))
+			if err != nil {
+				handle(msg)
+			}
+			decompressedData, err := ioutil.ReadAll(reader)
+			if err != nil {
+				handle(msg)
+			}
+			err = reader.Close()
+			msg.Data = decompressedData
+		}
+		handle(msg)
+	})
+	if err != nil {
+		//g.Log().Error(context.Background(), err)
+		log.Debug("error", err)
+	}
+}
+
+//
+//// 队列负载分组消费压缩消息
+//func (j *Jnats) QueueSubZip(sub, queue string, handle func(msg *nats.Msg)) {
+//	_, err := j.Nc.QueueSubscribe(sub, queue, func(msg *nats.Msg) {
+//		v := msg.Data
+//		if len(v) > 0 {
+//			res, err := gcompress.UnZlib(v)
+//			if err != nil {
+//				g.Log().Error(context.Background(), "NATS gcompress error: %v", err, msg)
+//			} else {
+//				msg.Data = res
+//			}
+//		}
+//		handle(msg)
+//	})
+//	if err != nil {
+//		g.Log().Error(context.Background(), err)
+//	}
+//}
+//
+////queue 队列消息暂时不用
+//
+//// 回复机制
+//// 生产压缩消息并请求回执
+//func (j *Jnats) PubReqZip(sub string, msg any, timeout time.Duration) (*nats.Msg, error) {
+//	res, err := gcompress.Zlib(g.NewVar(msg).Bytes())
+//	if err != nil {
+//		return nil, err
+//	} else {
+//		return j.Nc.Request(sub, res, timeout)
+//	}
+//}

src/jynats/jnatsjs/jnatsjs.go

@@ -0,0 +1,194 @@
+package jnatsjs
+
+//
+//import (
+//	"context"
+//	"github.com/gogf/gf/v2/encoding/gcompress"
+//	"jygit.jydev.jianyu360.cn/BP/jynats/jnats"
+//	"time"
+//
+//	"github.com/gogf/gf/v2/frame/g"
+//	"github.com/nats-io/nats.go"
+//)
+//
+//type Jnatsjs struct {
+//	Jnats *jnats.Jnats
+//	Jctx  nats.JetStreamContext
+//}
+//
+//func NewJnatsjs(addr string) *Jnatsjs {
+//	jn := jnats.NewJnats(addr)
+//	jctx, err := jn.Nc.JetStream()
+//	if err != nil {
+//		g.Log().Error(context.Background(), err)
+//	}
+//	jsn := &Jnatsjs{Jnats: jn, Jctx: jctx}
+//	jsn.NewDefaultStream()
+//	//jsn.AddDefaultConsumer()
+//	return jsn
+//}
+//
+//func (j *Jnatsjs) ReConnect() {
+//	if j.Jnats.ReConnect() {
+//		jctx, err := j.Jnats.Nc.JetStream()
+//		if err != nil {
+//			g.Log().Error(context.Background(), err)
+//		} else {
+//			j.Jctx = jctx
+//			j.NewDefaultStream()
+//		}
+//	}
+//}
+//
+//// 添加流，如果存在则不处理
+//func (j *Jnatsjs) NewStream(cfg *nats.StreamConfig) (*nats.StreamInfo, error) {
+//	st, err := j.Jctx.StreamInfo(cfg.Name)
+//	if st != nil {
+//		return st, err
+//	}
+//	return j.Jctx.AddStream(cfg)
+//}
+//
+//// 添加默认的流 jydef ,管理以jydef开头的主题
+//func (j *Jnatsjs) NewDefaultStream() (*nats.StreamInfo, error) {
+//	cfg := &nats.StreamConfig{
+//		Name:         "jydef",
+//		Subjects:     []string{"jydef.>"},
+//		Storage:      nats.MemoryStorage, // 儲存的方式 (預設 FileStorage)
+//		Retention:    nats.LimitsPolicy,  // 保留的策略
+//		Discard:      nats.DiscardOld,
+//		MaxAge:       72 * time.Hour, //保存三天
+//		MaxBytes:     2147483648,     //占用2G
+//		MaxMsgs:      2000000,        //二佰万消息
+//		MaxConsumers: 256,            //消费者
+//	}
+//	st, err := j.Jctx.StreamInfo(cfg.Name)
+//	if st != nil {
+//		return st, err
+//	}
+//	return j.Jctx.AddStream(cfg)
+//}
+//
+//// 添加默认的消费者
+//func (j *Jnatsjs) AddDefaultConsumer() (*nats.ConsumerInfo, error) {
+//	c, err := j.Jctx.ConsumerInfo("jydef", "jydef")
+//	if c != nil { //存在
+//		return c, err
+//	} else {
+//		if err != nil { //nats: consumer not found
+//			g.Log().Error(context.Background(), err)
+//		}
+//
+//		cfg := &nats.ConsumerConfig{
+//			Name:      "jydef",
+//			Durable:   "jydef",
+//			AckPolicy: nats.AckExplicitPolicy,
+//			AckWait:   3600 * time.Second,
+//
+//			MaxWaiting:    2048,
+//			MaxAckPending: 32,
+//		}
+//		c, err = j.Jctx.AddConsumer("jydef", cfg)
+//
+//	}
+//	return c, err
+//}
+//
+//// 发布消息
+//func (j *Jnatsjs) Pub(sub string, msg any) (nats.PubAckFuture, error) {
+//START:
+//	na, err := j.Jctx.PublishAsync(sub, g.NewVar(msg).Bytes())
+//	if err != nil { //nats: connection closed
+//		g.Log().Error(context.Background(), err)
+//		time.Sleep(220 * time.Millisecond)
+//		j.ReConnect()
+//		goto START
+//	}
+//	return na, err
+//
+//}
+//
+//// 发布压缩消息
+//func (j *Jnatsjs) PubZip(sub string, msg any) (nats.PubAckFuture, error) {
+//START:
+//	res, err := gcompress.Zlib(g.NewVar(msg).Bytes())
+//	if err != nil {
+//		return nil, err
+//	} else {
+//		na, err := j.Jctx.PublishAsync(sub, res)
+//		if err != nil { //nats: connection closed
+//			g.Log().Error(context.Background(), err)
+//			time.Sleep(220 * time.Millisecond)
+//			j.ReConnect()
+//			goto START
+//		}
+//		return na, err
+//	}
+//}
+//
+//// 直接消费消息,需要手动ack
+//func (j *Jnatsjs) Sub(sub, dur string, handle func(msg *nats.Msg)) {
+//	ctx := context.Background()
+//START:
+//	sub1, err1 := j.Jctx.PullSubscribe(sub, dur) //, nats.PullMaxWaiting(128)),配置为2048
+//	if err1 != nil {
+//		g.Log().Error(ctx, err1)
+//	}
+//	for {
+//		ctx1, _ := context.WithTimeout(context.Background(), 1*time.Second)
+//		msgs, err := sub1.Fetch(64, nats.Context(ctx1))
+//		if err == nil {
+//			for _, v := range msgs {
+//				handle(v)
+//			}
+//		} else {
+//			if err != context.DeadlineExceeded { //服务出问题,invalid subscription
+//				g.Log().Error(ctx, err)
+//				time.Sleep(200 * time.Millisecond)
+//				j.ReConnect()
+//				goto START
+//			}
+//		}
+//	}
+//	// START:
+//	// 	_, err := j.Jctx.Subscribe(sub, handle, nats.ManualAck(), nats.Durable(dur))
+//	// 	if err != nil {
+//	// 		g.Log().Error(context.Background(), err)
+//	// 		time.Sleep(200 * time.Millisecond)
+//	// 		j.ReConnect()
+//	// 		goto START
+//	// 	}
+//
+//}
+//
+//// 消费压缩消息，需要手动ack
+//func (j *Jnatsjs) SubZip(sub, dur string, handle func(msg *nats.Msg)) {
+//	ctx := context.Background()
+//START:
+//	sub1, err1 := j.Jctx.PullSubscribe(sub, dur) //, nats.PullMaxWaiting(128)),配置为2048
+//	if err1 != nil {
+//		g.Log().Error(ctx, err1)
+//	}
+//	for {
+//		ctx1, _ := context.WithTimeout(context.Background(), 1*time.Second)
+//		msgs, err := sub1.Fetch(64, nats.Context(ctx1))
+//		if err == nil {
+//			for _, v := range msgs {
+//				res, err := gcompress.UnZlib(v.Data)
+//				if err != nil {
+//					g.Log().Error(context.Background(), "NATS gcompress error: %v", err, v)
+//				} else {
+//					v.Data = res
+//				}
+//				handle(v)
+//			}
+//		} else {
+//			if err != context.DeadlineExceeded { //服务出问题,invalid subscription
+//				g.Log().Error(ctx, err)
+//				time.Sleep(200 * time.Millisecond)
+//				j.ReConnect()
+//				goto START
+//			}
+//		}
+//	}
+//}

src/jynats/main.go

@@ -0,0 +1,65 @@
+package main
+
+//
+//import (
+//	"fmt"
+//	"github.com/nats-io/nats.go"
+//	"jygit.jydev.jianyu360.cn/BP/jynats/jnats"
+//	"jygit.jydev.jianyu360.cn/BP/jynats/jnatsjs"
+//	"log"
+//	"time"
+//)
+//
+//func main() {
+//	var addr = "192.168.3.207:4223"
+//
+//	//一、无缓冲机制,使用jnats包
+//	jn := jnats.NewJnats(addr)
+//
+//	//先消费，带压缩
+//	jn.SubZip("test", func(msg *nats.Msg) {
+//		log.Println(string(msg.Data))
+//		//回执消息
+//		msg.Respond([]byte("receive msg:" + string(msg.Data)))
+//	})
+//	//后发布，带压缩
+//	go func() {
+//		for i := 0; i < 1; i++ {
+//			rep, err := jn.PubReqZip("test", fmt.Sprintf("你好，ping%d", i), time.Second)
+//			if err != nil {
+//				log.Println(err) //消息重试
+//			} else {
+//				log.Println(string(rep.Data))
+//			}
+//			time.Sleep(500 * time.Millisecond)
+//		}
+//	}()
+//
+//	//二、有缓存机制 使用jnatsjs包，重点：****默认流jydef的主题是以jydef开头，否则需要新建流****
+//	jsn := jnatsjs.NewJnatsjs(addr)
+//	go func() {
+//		for i := 0; i < 1000000; i++ {
+//			jsn.PubZip("jydef.test1", fmt.Sprintf("你好，测试%d", i))
+//			time.Sleep(1500 * time.Millisecond)
+//		}
+//	}()
+//	jsn.SubZip("jydef.test1", "aa1", func(msg *nats.Msg) {
+//		msg.Ack()
+//		log.Println(string(msg.Data))
+//	})
+//
+//	//
+//
+//	go func() {
+//		for i := 0; i < 1000000; i++ {
+//			jsn.Pub("jydef.test2", fmt.Sprintf("hello,%d", i))
+//			time.Sleep(1500 * time.Millisecond)
+//		}
+//	}()
+//	jsn.Sub("jydef.test2", "aa2", func(msg *nats.Msg) {
+//		msg.Ack()
+//		log.Println(string(msg.Data))
+//	})
+//
+//	select {}
+//}