From a99f2398af9db9dd7aa58e64314945e0b8bf18d8 Mon Sep 17 00:00:00 2001 From: lxy <237809796@qq.com> Date: Thu, 14 Jan 2021 11:37:36 +0000 Subject: [PATCH] =?UTF-8?q?=E5=8D=87=E7=BA=A7=EF=BC=9Aztimer=20=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E5=91=A8=E6=9C=9F=E5=BE=AA=E7=8E=AF=E5=AE=9A=E6=97=B6?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E8=A1=A8=E8=BE=BE=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit git-svn-id: svn://47.119.165.148/zhub@70 e63fbceb-bcc3-4977-ac22-735b83d8d0f4 --- cli_test.go | 22 ++++++++++++++--- pkg.bat | 3 +++ zsub/zsub.go | 2 +- zsub/ztimer.go | 66 ++++++++++++++++++++++++++++++++++++++------------ 4 files changed, 73 insertions(+), 20 deletions(-) diff --git a/cli_test.go b/cli_test.go index 97ae24c..b5a06d7 100644 --- a/cli_test.go +++ b/cli_test.go @@ -48,20 +48,36 @@ func TestCli(t *testing.T) { func TestTimer(t *testing.T) { go func() { - client, _ := cli.Create(addr, "topic-x") + client, _ := cli.Create(addr, "topic-1") client.Timer("a", func() { log.Println("client-1 收到 a 的定时消息") }) }() - time.Sleep(time.Second * 5) go func() { - client, _ := cli.Create(addr, "topic-x") + client, _ := cli.Create(addr, "topic-2") client.Timer("a", func() { log.Println("client-2 收到 a 的定时消息") }) }() + go func() { + client, _ := cli.Create(addr, "topic-3") + client.Timer("c", func() { + log.Println("client-2 收到 c 的定时消息") + }) + + client.Timer("b", func() { + log.Println("client-2 收到 b 的定时消息") + }) + client.Timer("STANDING-DOWNLOAD-GAME", func() { + log.Println("client-2 收到 STANDING-DOWNLOAD-GAME 的定时消息") + }) + client.Timer("VIP-EXP-EXPIRE", func() { + log.Println("client-2 收到 VIP-EXP-EXPIRE 的定时消息") + }) + }() + time.Sleep(time.Hour * 3) } diff --git a/pkg.bat b/pkg.bat index 543aceb..f3d8cc3 100644 --- a/pkg.bat +++ b/pkg.bat @@ -2,3 +2,6 @@ SET GOOS=linux SET GOARCH=amd64 go build -o zhub.sh -ldflags "-s -w" ./main.go upx -9 zhub.sh + +scp zhub.sh zhost: +del zhub.sh diff --git a/zsub/zsub.go b/zsub/zsub.go index 1b2af0f..262de90 100644 --- a/zsub/zsub.go +++ b/zsub/zsub.go @@ -167,7 +167,7 @@ func ServerStart(host string, port int) { log.Println(err) continue } - fmt.Println("conn start: ", conn.RemoteAddr()) + log.Println("conn start: ", conn.RemoteAddr()) go zsub.acceptHandler(&ZConn{ conn: &conn, diff --git a/zsub/ztimer.go b/zsub/ztimer.go index e45fa0c..9f52b7e 100644 --- a/zsub/ztimer.go +++ b/zsub/ztimer.go @@ -8,8 +8,11 @@ import ( "github.com/robfig/cron" "log" "os/exec" + "regexp" + "strconv" "strings" "text/template" + "time" ) type ZTimer struct { @@ -17,6 +20,7 @@ type ZTimer struct { expr string topic string cron *cron.Cron + ticker *time.Ticker single bool } @@ -38,30 +42,60 @@ func (s *ZSub) timer(rcmd []string, c *ZConn) { timer.conns = c.appendTo(timer.conns) } - // todo: when timer.expr changed send message to all the timer‘s subscribe if len(rcmd) == 4 && !strings.EqualFold(timer.expr, rcmd[2]) { timer.expr = rcmd[2] if timer.cron != nil { timer.cron.Stop() } - timer.cron = func() *cron.Cron { - c := cron.New() - c.AddFunc(timer.expr, func() { - for _, conn := range timer.conns { - err := send(conn.conn, "timer", timer.topic) - if timer.single && err == nil { - break - } + if timer.ticker != nil { + timer.ticker.Stop() + } + + var timerFun = func() { + for _, conn := range timer.conns { + err := send(conn.conn, "timer", timer.topic) + if timer.single && err == nil { + break } - }) - go c.Run() - return c - }() - timer.configSave() + } + } + + r, _ := regexp.Compile("^\\d+[d,H,m,s]$") + expr := timer.expr + if r.MatchString(expr) { + n, _ := strconv.Atoi(expr[:len(expr)-1]) + _n := time.Duration(n) + var ticker *time.Ticker + switch expr[len(expr)-1:] { + case "d": + ticker = time.NewTicker(_n * time.Hour * 24) + case "H": + ticker = time.NewTicker(_n * time.Hour) + case "m": + ticker = time.NewTicker(_n * time.Minute) + case "s": + ticker = time.NewTicker(_n * time.Second) + } + + timer.ticker = ticker + go func() { + for range ticker.C { + timerFun() + } + }() + } else { + timer.cron = func() *cron.Cron { + c := cron.New() + c.AddFunc(timer.expr, timerFun) + go c.Run() + return c + }() + } + //timer.configSave() } if len(rcmd) == 4 && (strings.EqualFold("a", rcmd[3]) != timer.single) { timer.single = strings.EqualFold("a", rcmd[3]) - timer.configSave() + //timer.configSave() } s.timers[rcmd[1]] = timer @@ -134,7 +168,7 @@ func executeShell(command string) (string, error, string) { func (s *ZSub) reloadTimerConfig() { db, err := sql.Open("mysql", "root:*Zhong123098!@tcp(47.111.150.118:6063)/platf_oth?charset=utf8") // dev - //db, err := sql.Open("mysql", "root:*Hello@27.com!@tcp(0.0.0.0:6033)/platf_oth?charset=utf8") // qc + //db, err := sql.Open("mysql", "root:*Zhong123098!@tcp(121.196.17.55:6063)/platf_oth?charset=utf8") // qc //db, err := sql.Open("mysql", "root:*Hello@27.com!@tcp(122.112.180.156:6033)/platf_oth?charset=utf8") // pro if err != nil { log.Println(err)