备份清洗 。。。 调整修补 。。。

sensitive/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
+

sensitive/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

sensitive/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>

sensitive/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>

sensitive/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>

sensitive/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 = []
+}
+
+```

sensitive/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
+)

sensitive/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=

sensitive/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)
+//	}
+//}

sensitive/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
+//			}
+//		}
+//	}
+//}

sensitive/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 {}
+//}

sensitive/src/main.go

@@ -10,8 +10,8 @@ func init() {
 	ul.InitC()
 }
 func main() {
-	//go ul.AddTaskSensitiveWordsData() //增量-新增
-	//ul.SentiveUdp() //数据流程-增量
+	go ul.AddTaskSensitiveWordsData() //增量-新增
+	ul.SentiveUdp()                   //数据流程-增量
 	//ul.TestData() //临时测试数据
 
 	lock := make(chan bool)

sensitive/src/util/udpdata.go

@@ -47,6 +47,13 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 		} else {
 			sid, _ := rep["gtid"].(string)
 			eid, _ := rep["lteid"].(string)
+			stype := qu.ObjToString(rep["stype"])
+			if stype == "monitor" {
+				log.Println("收到监测......")
+				key := qu.ObjToString(rep["key"])
+				udpclient.WriteUdp([]byte(key), mu.OP_NOOP, ra)
+				return
+			}
 			if sid == "" || eid == "" {
 				log.Println("err", "sid=", sid, ",eid=", eid)
 				return
@@ -56,7 +63,6 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 			QuerySensitiveWords(sid, eid)
 			log.Println("...计划发送udp~统计下一节点...")
 			//
-
 		}
 	case mu.OP_NOOP: //下个节点回应
 		log.Println(string(data))

src/main.go

@@ -41,20 +41,21 @@ var (
 	S_ProvinceDict                        map[string][]S_Province //省份-map
 	S_CityDict                            map[string][]S_City     //城市-map
 	S_DistrictDict                        map[string][]S_District //区县-map
-
 	//删除字段
 	unset_dict          = map[string]interface{}{"winner": 1, "s_winner": 1, "bidamount": 1, "winnerorder": 1}
 	udplock, getasklock sync.Mutex
 	taskList            []map[string]interface{}
+	//监控相关
+	responselock     sync.Mutex
+	lastNodeResponse int64
 )
 
-//初始化城市
+// 初始化城市
 func initCheckCity() {
 	//初始化-城市配置
 	S_ProvinceDict = make(map[string][]S_Province, 0)
 	S_CityDict = make(map[string][]S_City, 0)
 	S_DistrictDict = make(map[string][]S_District, 0)
-
 	q := map[string]interface{}{
 		"town_code": map[string]interface{}{
 			"$exists": 0,
@@ -111,7 +112,7 @@ func initCheckCity() {
 	log.Println(fmt.Sprintf("城市配置加载完毕...省~%d 市~%d 区~%d", len(S_ProvinceDict), len(S_CityDict), len(S_DistrictDict)))
 }
 
-//mgo-配置等
+// mgo-配置等
 func initMgo() {
 	mconf = Sysconfig["mongodb"].(map[string]interface{})
 	log.Println(mconf)
@@ -133,7 +134,7 @@ func initMgo() {
 	qy_mgo.InitPool()
 
 	bid_mgo = &MongodbSim{
-		MongodbAddr: "172.17.145.163:27083,172.17.4.187:27082",
+		MongodbAddr: "172.17.189.140:27080,172.17.189.141:27081",
 		DbName:      "qfw",
 		Size:        10,
 		UserName:    "zhengkun",
@@ -151,7 +152,7 @@ func initMgo() {
 	log.Println("mgo 等配置，加载完毕...")
 }
 
-//初始化
+// 初始化
 func init() {
 	qu.ReadConfig(&Sysconfig) //加载配置文件
 	log.Println(Sysconfig)
@@ -167,26 +168,21 @@ func init() {
 }
 
 func main() {
+	lastNodeResponse = time.Now().Unix()
 	updport := Sysconfig["udpport"].(string)
 	udpclient = mu.UdpClient{Local: updport, BufSize: 1024}
 	udpclient.Listen(processUdpMsg)
 	log.Println("Udp服务监听", updport)
 
 	go getRepeatTask()
+	go checkMailJob()
+	go lastUdpJob()
 
 	lock := make(chan bool)
 	<-lock
 }
 
-//临时校验
-func mainT() {
-	sid := "12982d658fa2ac55ba96517d"
-	eid := "92982d658fa2ac55ba96517e"
-	testCheckData(sid, eid)
-	time.Sleep(99999 * time.Hour)
-}
-
-//开始审查数据
+// 开始审查数据
 func startCheckData(sid, eid string) {
 	defer qu.Catch()
 	q := map[string]interface{}{
@@ -215,26 +211,16 @@ func startCheckData(sid, eid string) {
 			}()
 			//更新-
 			update_check := make(map[string]interface{}, 0)
-
-			//审查-城市
-			//getCheckDataCity(tmp, &update_check)
-			//审查-中标金额
-			//getCheckDataBidamount(tmp, &update_check)
-			//验证是否修复发布时间 - 对比开标日期，投标截止日期
-			getCheckDataPublishtime(tmp, &update_check)
-			//清洗分类~
-			//is_category:= getCheckDataCategory(tmp,&update_check)
-
+			//getCheckDataCity(tmp, &update_check)		//审查-城市
+			//getCheckDataBidamount(tmp, &update_check)	//审查-中标金额
+			getCheckDataPublishtime(tmp, &update_check) //修复-发布时间
+			//getCheckDataCategory(tmp,&update_check)	//修复分类
 			//最终计算是否清洗
 			update_dict := make(map[string]interface{}, 0)
 			if len(update_check) > 0 {
 				update_dict["$set"] = update_check
 			}
-			//if is_category {
-			//	update_dict["$unset"] = unset_dict
-			//}
-			if len(update_dict) > 0 {
-				//注意事项~更新key 不能与 删除key  同时存在
+			if len(update_dict) > 0 { //注意事项~更新key不能与删除key同时存在
 				isRepair++
 				UpdateTask.updatePool <- []map[string]interface{}{
 					update_id,
@@ -245,13 +231,11 @@ func startCheckData(sid, eid string) {
 		tmp = make(map[string]interface{})
 	}
 	check_wg.Wait()
-
 	log.Println("data_clean is over ", total, "~", isRepair)
-
 	sendNextNode(sid, eid)
 }
 
-//udp监听
+// udp监听
 func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 	switch act {
 	case mu.OP_TYPE_DATA:
@@ -262,68 +246,83 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 		} else {
 			sid, _ := rep["gtid"].(string)
 			eid, _ := rep["lteid"].(string)
+			stype := qu.ObjToString(rep["stype"])
+			key := qu.ObjToString(rep["key"])
+			if stype == "monitor" {
+				log.Println("收到监测......")
+				udpclient.WriteUdp([]byte(key), mu.OP_NOOP, ra)
+				return
+			}
 			if sid == "" || eid == "" {
 				log.Println("err", "sid=", sid, ",eid=", eid)
 				return
 			} else {
-				go udpclient.WriteUdp(data, mu.OP_NOOP, ra)
-				//插入任务
+				lastNodeResponse = time.Now().Unix()
+				udpclient.WriteUdp([]byte(key), mu.OP_NOOP, ra)
 				udplock.Lock()
 				taskList = append(taskList, map[string]interface{}{
 					"sid": sid,
 					"eid": eid,
-				})
+				}) //插入任务
 				log.Println("udp收到任务...数量：", len(taskList), "具体任务：", taskList)
 				udplock.Unlock()
-
 			}
 		}
 	case mu.OP_NOOP: //下个节点回应
-		log.Println("下节点回应", string(data))
+		log.Println("下节点回应:", string(data))
+		udptaskmap.Delete(string(data))
 	}
 }
 
-//发送下阶段节点~
+// 发送下阶段节点~
 func sendNextNode(sid string, eid string) {
 	//更新记录状态
 	updateProcessUdpIdsInfo(sid, eid)
-
-	for _, m := range nextNode {
+	log.Println("判重任务完成...发送下节点udp...")
+	for _, to := range nextNode {
+		key := sid + "-" + eid + "-" + qu.ObjToString(to["stype"])
 		by, _ := json.Marshal(map[string]interface{}{
 			"gtid":  sid,
 			"lteid": eid,
-			"stype": qu.ObjToString(m["stype"]),
+			"stype": qu.ObjToString(to["stype"]),
+			"key":   key,
 		})
-		new_err := udpclient.WriteUdp(by, mu.OP_TYPE_DATA, &net.UDPAddr{
-			IP:   net.ParseIP(m["addr"].(string)),
-			Port: qu.IntAll(m["port"]),
-		})
-		if new_err != nil {
-			log.Println(new_err)
+		addr := &net.UDPAddr{
+			IP:   net.ParseIP(to["addr"].(string)),
+			Port: qu.IntAll(to["port"]),
 		}
+		node := &udpNode{by, addr, time.Now().Unix(), 0}
+		udptaskmap.Store(key, node)
+		udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
 	}
-	log.Println("udp通知审查数据完成,通知下节点")
 }
 
-//更新流程记录id段落
+// 更新流程记录id段落
 func updateProcessUdpIdsInfo(sid string, eid string) {
 	query := map[string]interface{}{
-		"gtid":  sid,
-		"lteid": eid,
+		"gtid": map[string]interface{}{
+			"$gte": sid,
+		},
+		"lteid": map[string]interface{}{
+			"$lte": eid,
+		},
 	}
-	log.Println("开始更新流程段落记录~~", query)
-	data := bid_mgo.FindOne("bidding_processing_ids", query)
-	if len(data) > 0 {
-		up_id := BsonTOStringId(data["_id"])
-		if up_id != "" {
-			update := map[string]interface{}{
-				"$set": map[string]interface{}{
-					"dataprocess": 4,
-					"updatetime":  time.Now().Unix(),
-				},
+	task_coll := "bidding_processing_ids"
+	datas, _ := bid_mgo.Find(task_coll, query, nil, nil)
+	if len(datas) > 0 {
+		log.Println("开始更新流程段落记录~~", len(datas), "段")
+		for _, v := range datas {
+			up_id := BsonTOStringId(v["_id"])
+			if up_id != "" {
+				update := map[string]interface{}{
+					"$set": map[string]interface{}{
+						"dataprocess": 4,
+						"updatetime":  time.Now().Unix(),
+					},
+				}
+				bid_mgo.UpdateById(task_coll, up_id, update)
+				log.Println("流程段落记录~~更新完毕~", update)
 			}
-			bid_mgo.UpdateById("bidding_processing_ids", up_id, update)
-			log.Println("流程段落记录~~更新完毕~", update)
 		}
 	} else {
 		log.Println("未查询到记录id段落~", query)
@@ -351,76 +350,55 @@ func httpDo(detail string) (e error) {
 	return nil
 }
 
-//监听-获取-分发清洗任务
+// 监听-获取-分发清洗任务
 func getRepeatTask() {
 	for {
 		if len(taskList) > 0 {
 			getasklock.Lock()
-			mapInfo := taskList[0]
-			if mapInfo != nil {
-				taskList = taskList[1:]
-				log.Println("获取任务段处理中~~~剩余任务池~~~", len(taskList), taskList)
-				sid := qu.ObjToString(mapInfo["sid"])
-				eid := qu.ObjToString(mapInfo["eid"])
-				startCheckData(sid, eid)
+			len_list := len(taskList)
+			if len_list > 1 {
+				first_id := qu.ObjToString(taskList[0]["sid"])
+				end_id := qu.ObjToString(taskList[len_list-1]["eid"])
+				if first_id != "" && end_id != "" {
+					taskList = taskList[len_list:]
+					log.Println("合并段落~正常~", first_id, "~", end_id, "~剩余任务池~", len(taskList), taskList)
+					startCheckData(first_id, end_id)
+				} else {
+					log.Println("合并段落~错误~正常取段落~~~")
+					mapInfo := taskList[0]
+					if mapInfo != nil {
+						taskList = taskList[1:]
+						log.Println("获取任务段处理中~~~剩余任务池~~~", len(taskList), taskList)
+						sid := qu.ObjToString(mapInfo["sid"])
+						eid := qu.ObjToString(mapInfo["eid"])
+						startCheckData(sid, eid)
+					}
+				}
+			} else {
+				mapInfo := taskList[0]
+				if mapInfo != nil {
+					taskList = taskList[1:]
+					log.Println("获取任务段处理中~~~剩余任务池~~~", len(taskList), taskList)
+					sid := qu.ObjToString(mapInfo["sid"])
+					eid := qu.ObjToString(mapInfo["eid"])
+					startCheckData(sid, eid)
+				}
 			}
 			getasklock.Unlock()
 		} else {
-			time.Sleep(15 * time.Second)
+			time.Sleep(10 * time.Second)
 		}
 	}
 }
 
-//测试版
-func testCheckData(sid, eid string) {
-	defer qu.Catch()
-	q := map[string]interface{}{
-		"_id": map[string]interface{}{
-			"$gt":  StringTOBsonId(sid),
-			"$lte": StringTOBsonId(eid),
-		},
-	}
-	check_pool := make(chan bool, check_thread)
-	check_wg := &sync.WaitGroup{}
-	sess := data_mgo.GetMgoConn()
-	defer data_mgo.DestoryMongoConn(sess)
-	it := sess.DB(data_mgo.DbName).C(coll_name).Find(&q).Iter()
-	total, isRepair := 0, 0
-	for tmp := make(map[string]interface{}); it.Next(&tmp); total++ {
-		if total%10000 == 0 {
-			log.Println("当前数量:", total, isRepair, tmp["_id"])
+func lastUdpJob() {
+	for {
+		responselock.Lock()
+		if time.Now().Unix()-lastNodeResponse >= 1800 {
+			lastNodeResponse = time.Now().Unix() //重置时间
+			sendErrMailApi("数据清洗~发现处理流程超时~给予告警", fmt.Sprintf("半小时左右~无新段落数据进入清洗增量流程...相关人员检查..."))
 		}
-		update_id := map[string]interface{}{"_id": tmp["_id"]}
-		check_pool <- true
-		check_wg.Add(1)
-		go func(tmp map[string]interface{}, update_id map[string]interface{}) {
-			defer func() {
-				<-check_pool
-				check_wg.Done()
-			}()
-			//更新-
-			update_check := make(map[string]interface{}, 0)
-
-			//审查-城市
-			getCheckDataCity(tmp, &update_check)
-			//最终计算是否清洗
-			update_dict := make(map[string]interface{}, 0)
-			if len(update_check) > 0 {
-				update_check["is_standard"] = 1
-				update_dict["$set"] = update_check
-			}
-			if len(update_dict) > 0 {
-				//注意事项~更新key 不能与 删除key  同时存在
-				isRepair++
-				UpdateTask.updatePool <- []map[string]interface{}{
-					update_id,
-					update_dict,
-				}
-			}
-		}(tmp, update_id)
-		tmp = make(map[string]interface{})
+		responselock.Unlock()
+		time.Sleep(300 * time.Second)
 	}
-	check_wg.Wait()
-
-	log.Println("data_clean is over ", total, "~", isRepair)
 }

src/udptaskmap.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net"
+	"net/http"
+	"sync"
+	"time"
+)
+
+var udptaskmap = &sync.Map{}
+var tomail string
+var api string
+
+type udpNode struct {
+	data      []byte
+	addr      *net.UDPAddr
+	timestamp int64
+	retry     int
+}
+
+func checkMailJob() {
+	jkmail, _ := Sysconfig["jkmail"].(map[string]interface{})
+	if jkmail != nil {
+		tomail, _ = jkmail["to"].(string)
+		api, _ = jkmail["api"].(string)
+	}
+	log.Println("start checkMailJob", tomail, Sysconfig["jkmail"])
+	for {
+		udptaskmap.Range(func(k, v interface{}) bool {
+			now := time.Now().Unix()
+			node, _ := v.(*udpNode)
+			if now-node.timestamp > 120 {
+				udptaskmap.Delete(k)
+				info_str := fmt.Sprintf("下节点~行业分类~未响应~相关人员检查~%s", k.(string))
+				res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", api, tomail, "增量判重程序~严重警告", info_str))
+				if err == nil {
+					defer res.Body.Close()
+					read, err := ioutil.ReadAll(res.Body)
+					log.Println("邮件发送：", string(read), err)
+				}
+			}
+			return true
+		})
+		time.Sleep(60 * time.Second)
+	}
+}
+
+func sendErrMailApi(title, body string) {
+	jkmail, _ := Sysconfig["jkmail"].(map[string]interface{})
+	if jkmail != nil {
+		tomail, _ = jkmail["to"].(string)
+		api, _ = jkmail["api"].(string)
+	}
+	log.Println(tomail, api)
+	res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", api, tomail, title, body))
+	if err == nil {
+		defer res.Body.Close()
+		read, err := ioutil.ReadAll(res.Body)
+		log.Println("邮件发送成功：", string(read), err)
+	} else {
+		log.Println("邮件发送失败：", err)
+	}
+}