diff --git a/.gitignore b/.gitignore index 87c5d9d7..fadb7396 100644 --- a/.gitignore +++ b/.gitignore @@ -14,4 +14,5 @@ # Dependency directories (remove the comment below to include it) vendor/ .idea -_backup \ No newline at end of file +_backup +_example \ No newline at end of file diff --git a/crontab/README.md b/crontab/README.md index b270da35..e331aa0b 100644 --- a/crontab/README.md +++ b/crontab/README.md @@ -6,7 +6,7 @@ package main import ( - "fmt" + "log" "github.com/go-kratos/kratos/v2" "github.com/redis/go-redis/v9" @@ -17,35 +17,35 @@ import ( ) func main() { - rdb := redis.NewClient(&redis.Options{ - Addr: "127.0.0.1:6379", - }) - - if err := kratos.New( - kratos.Server( - NewCrontabServer(rdb), - ), - ).Run(); err != nil { - panic(err) - } -} - -func NewCrontabServer(rdb redis.Cmdable) *crontab.Server { c := cron.New( cron.WithSeconds(), ) - c.AddFunc("*/1 * * * * *", func() { - fmt.Println("Every hour on the half hour") + c.AddFunc("* * * * * *", func() { + log.Println("Hello world") }) - return crontab.NewServer( - c, - crontab.WithName("crontab:server"), - crontab.WithDebug(), - crontab.WithMutex( - redisMutex.New(rdb), + app := kratos.New( + kratos.Server( + crontab.NewServer(c, + crontab.WithMutex(redisMutex.New(redis.NewClient(&redis.Options{ + Addr: "localhost:6379", + }))), + crontab.WithDebug(), + ), ), ) + + app.Run() } +``` + +output: + +```bash +2023/12/25 14:25:56 crontab: server started +2023/12/25 14:25:57 Hello world +2023/12/25 14:25:58 Hello world +2023/12/25 14:25:59 Hello world +2023/12/25 14:26:00 Hello world ``` \ No newline at end of file diff --git a/crontab/server.go b/crontab/server.go index ad096f64..0ea2a59c 100644 --- a/crontab/server.go +++ b/crontab/server.go @@ -65,15 +65,14 @@ func NewServer(c *cron.Cron, opts ...Option) *Server { } func (s *Server) Start(ctx context.Context) error { - go s.run(ctx) - - return nil -} - -func (s *Server) run(ctx context.Context) { timer := time.NewTicker(time.Second) + defer func() { - _ = s.mutex.Unlock(ctx, s.name) + if s.running { + s.cron.Stop() + } + + s.mutex.Unlock(ctx, s.name) timer.Stop() }() @@ -81,22 +80,22 @@ func (s *Server) run(ctx context.Context) { select { case <-ctx.Done(): s.log("crontab: server done") - return + return ctx.Err() case <-s.stoped: s.log("crontab: server stoped") - return + return nil case <-timer.C: if err := s.mutex.Lock(ctx, s.name); err != nil { s.log(err) continue } - s.start(ctx) + s.start() } } } -func (s *Server) start(ctx context.Context) { +func (s *Server) start() { s.runningMu.Lock() defer s.runningMu.Unlock() @@ -111,19 +110,11 @@ func (s *Server) start(ctx context.Context) { } func (s *Server) Stop(ctx context.Context) error { - s.runningMu.Lock() - defer s.runningMu.Unlock() - - if !s.running { - return nil - } - - s.running = false - s.cron.Stop() + s.log("crontab: server stopping") close(s.stoped) - return s.mutex.Unlock(ctx, s.name) + return nil } func (s *Server) log(v ...interface{}) {