lxy/zhub
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

删除：topic 消息写数据库

Browse Source 
git-svn-id: svn://47.119.165.148/zhub@163 e63fbceb-bcc3-4977-ac22-735b83d8d0f4
This commit is contained in:
lxy
2023-03-07 06:50:53 +00:00
parent 71565c9a6c
commit c4709e2529
2 changed files with 59 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
zsub/msg-consumer.go
View File

@@ -80,9 +80,9 @@ func msgAccept(v Message) {
if len(rcmd) != 3 { if len(rcmd) != 3 {
c.send("-Error: publish para number![" + strings.Join(rcmd, " ") + "]") c.send("-Error: publish para number![" + strings.Join(rcmd, " ") + "]")
} else { } else {
if len(topicChan) < cap(topicChan) { /*if len(topicChan) < cap(topicChan) {
topicChan <- rcmd topicChan <- rcmd
} }*/
zsub.Publish(rcmd[1], rcmd[2]) zsub.Publish(rcmd[1], rcmd[2])
} }
return return
@@ -90,9 +90,9 @@ func msgAccept(v Message) {
if len(rcmd) != 3 { if len(rcmd) != 3 {
c.send("-Error: publish para number![" + strings.Join(rcmd, " ") + "]") c.send("-Error: publish para number![" + strings.Join(rcmd, " ") + "]")
} else { } else {
if len(topicChan) < cap(topicChan) { /*if len(topicChan) < cap(topicChan) {
topicChan <- rcmd topicChan <- rcmd
} }*/
zsub.Publish(rcmd[1], rcmd[2]) zsub.Publish(rcmd[1], rcmd[2])
} }
return return

107
zsub/zdb.go
View File

@@ -8,13 +8,13 @@ import (
"os" "os"
"strconv" "strconv"
"strings" "strings"
"sync/atomic"
"time" "time"
) )
/*
var ( var (
topicChan = make(chan []string, 1000) //接收到的 所有消息数据 topicChan = make(chan []string, 1000) //接收到的 所有消息数据, 用于写入数据库持久化
) )*/
// Message 数据封装 // Message 数据封装
type Message struct { type Message struct {
@@ -172,57 +172,60 @@ var (
func init() { func init() {
LoadConf("app.conf") LoadConf("app.conf")
_db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8", /*
GetStr("ztimer.db.user", "root"), _db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8",
GetStr("ztimer.db.pwd", "123456"), GetStr("ztimer.db.user", "root"),
GetStr("ztimer.db.addr", "127.0.0.1:3306"), GetStr("ztimer.db.pwd", "123456"),
GetStr("ztimer.db.database", "zhub"), GetStr("ztimer.db.addr", "127.0.0.1:3306"),
)) GetStr("ztimer.db.database", "zhub"),
if err != nil { ))
log.Println(err) if err != nil {
return log.Println(err)
} return
}
db = _db db = _db
// 批量写入数据库，等待超时5秒，如有数据写入数据 // 批量写入数据库，等待超时5秒，如有数据写入数据
go func() {
defer func() { go func() {
if r := recover(); r != nil { defer func() {
log.Println("MsgToDb Recovered:", r) if r := recover(); r != nil {
log.Println("MsgToDb Recovered:", r)
}
}()
var flagcount = 0
var _sql = "INSERT INTO zhub.topicmessage (`msgid`,`topic`,`value`,`createtime`) VALUES \n"
for {
select {
case msg := <-topicChan:
var topic, value = msg[1], msg[2]
var t = time.Now().UnixNano() / 1e6
_sql += fmt.Sprintf("('%s','%s','%s',%d),\n",
strconv.FormatInt(t, 36)+"-"+strconv.FormatInt(atomic.AddInt64(&seq, 1), 36), topic, value, t)
flagcount++
case <-time.After(time.Second * 5): // 等待5秒
if flagcount > 0 {
flagcount = 100
}
}
if flagcount != 100 {
continue
}
_sql = _sql[:len(_sql)-2]
_sql += ";"
_, err = db.Exec(_sql)
if err != nil {
log.Println(err)
}
_sql = "INSERT INTO zhub.topicmessage (`msgid`,`topic`,`value`,`createtime`) VALUES \n"
flagcount = 0
} }
}() }()
*/
var flagcount = 0
var _sql = "INSERT INTO zhub.topicmessage (`msgid`,`topic`,`value`,`createtime`) VALUES \n"
for {
select {
case msg := <-topicChan:
var topic, value = msg[1], msg[2]
var t = time.Now().UnixNano() / 1e6
_sql += fmt.Sprintf("('%s','%s','%s',%d),\n",
strconv.FormatInt(t, 36)+"-"+strconv.FormatInt(atomic.AddInt64(&seq, 1), 36), topic, value, t)
flagcount++
case <-time.After(time.Second * 5): // 等待5秒
if flagcount > 0 {
flagcount = 100
}
}
if flagcount != 100 {
continue
}
_sql = _sql[:len(_sql)-2]
_sql += ";"
_, err = db.Exec(_sql)
if err != nil {
log.Println(err)
}
_sql = "INSERT INTO zhub.topicmessage (`msgid`,`topic`,`value`,`createtime`) VALUES \n"
flagcount = 0
}
}()
} }