From f95054baa8bb28a3c82d2a9a71aeb66d783464c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E5=85=89=E6=98=A5?= Date: Sun, 12 Jun 2022 13:13:27 +0800 Subject: [PATCH] - update --- .gitignore | 3 +- etcd_server_test.go | 67 -------------------- etcd_worker_test.go | 70 --------------------- grpc_server_test.go | 147 -------------------------------------------- 4 files changed, 2 insertions(+), 285 deletions(-) delete mode 100644 etcd_server_test.go delete mode 100644 etcd_worker_test.go delete mode 100644 grpc_server_test.go diff --git a/.gitignore b/.gitignore index 4d088b9..09cbaef 100644 --- a/.gitignore +++ b/.gitignore @@ -8,4 +8,5 @@ gomod.sh /vendor/ grpc_build.sh etcd_server_test.go -etcd_worker_test.go \ No newline at end of file +etcd_worker_test.go +grpc_server_test.go \ No newline at end of file diff --git a/etcd_server_test.go b/etcd_server_test.go deleted file mode 100644 index 67b95c1..0000000 --- a/etcd_server_test.go +++ /dev/null @@ -1,67 +0,0 @@ -package gojobs - -import ( - "context" - "github.com/robfig/cron/v3" - "log" - "testing" - "time" -) - -func TestEtcdServer(t *testing.T) { - server, err := NewEtcdServer(&EtcdConfig{ - Endpoints: []string{"http://127.0.0.1:2379"}, - DialTimeout: time.Second * 5, - Username: "root", - Password: "p5sttPYcFWw7Z7aP", - }) - if err != nil { - panic(err) - } - defer server.Close() - - // 创建一个cron实例 精确到秒 - c := cron.New(cron.WithSeconds()) - - // 每隔15秒执行一次 - _, _ = c.AddFunc("*/15 * * * * *", func() { - - create, err := server.Create(context.TODO(), server.IssueWatchKey("116.30.228.12")+"/"+"wechat_1_test", "每隔15秒执行一次") - if err != nil { - log.Println("创建任务失败", err) - } - log.Println("创建任务成功", create, err) - - }) - - // 每隔30秒执行一次 - _, _ = c.AddFunc("*/30 * * * * *", func() { - - create, err := server.Create(context.TODO(), server.IssueWatchKey("127.0.0.1")+"/"+"wechat_2_test", "每隔30秒执行一次") - if err != nil { - log.Println("创建任务失败", err) - } - log.Println("创建任务成功", create, err) - - }) - - // 每隔1分钟执行一次 - _, _ = c.AddFunc("0 */1 * * * *", func() { - - create, err := server.Create(context.TODO(), server.IssueWatchKey("116.30.228.12")+"/"+"wechat_3_test", "每隔1分钟执行一次") - if err != nil { - log.Println("创建任务失败", err) - } - log.Println("创建任务成功", create, err) - - workers, _ := server.ListWorkers() - log.Println("ListWorkers", workers) - }) - - // 启动任务 - c.Start() - - // 关闭任务 - defer c.Stop() - select {} -} diff --git a/etcd_worker_test.go b/etcd_worker_test.go deleted file mode 100644 index bb07877..0000000 --- a/etcd_worker_test.go +++ /dev/null @@ -1,70 +0,0 @@ -package gojobs - -import ( - "context" - "fmt" - "go.etcd.io/etcd/api/v3/mvccpb" - "go.etcd.io/etcd/client/v3" - "log" - "sync" - "testing" - "time" -) - -func TestEtcdWorker(t *testing.T) { - server, err := NewEtcdWorker(&EtcdConfig{ - Endpoints: []string{"http://127.0.0.1:2379"}, - DialTimeout: time.Second * 5, - Username: "root", - Password: "p5sttPYcFWw7Z7aP", - }) - if err != nil { - panic(err) - } - defer server.Close() - - // 监听 - go func() { - rch := server.Watch(context.TODO(), server.GetWatchKey()+"/", clientv3.WithPrefix()) - // 处理监听事件 - for watchResp := range rch { - for _, watchEvent := range watchResp.Events { - switch watchEvent.Type { - case mvccpb.PUT: - - // 收到任务 - log.Printf("监听收到任务 键名:%s 值:%s\n", watchEvent.Kv.Key, watchEvent.Kv.Value) - log.Println("处理监听") - - wg := sync.WaitGroup{} - wg.Add(1) - go run(&wg) - wg.Wait() - - _, err = server.Delete(context.TODO(), string(watchEvent.Kv.Key)) - if err != nil { - log.Println("删除失败", err) - } else { - log.Println("删除成功") - } - case mvccpb.DELETE: - // 任务被删除了 - - } - - } - } - log.Println("out") - }() - - select {} -} - -func run(wg *sync.WaitGroup) { - log.Println("等待开始") - fmt.Println(time.Now()) - time.Sleep(time.Second * 10) - fmt.Println(time.Now()) - log.Println("等待结束") - wg.Done() -} diff --git a/grpc_server_test.go b/grpc_server_test.go deleted file mode 100644 index 032b2a2..0000000 --- a/grpc_server_test.go +++ /dev/null @@ -1,147 +0,0 @@ -package gojobs - -import ( - "github.com/robfig/cron/v3" - "go.dtapp.net/gojobs/pb" - "go.dtapp.net/gouuid" - "io" - "log" - "sync" - "testing" - "time" -) - -func TestJobs(t *testing.T) { - wg := sync.WaitGroup{} - wg.Add(4) - go testServer(&wg) - go testCron(&wg) - go testWorker1(&wg) - go testWorker2(&wg) - wg.Wait() -} - -func testServer(wg *sync.WaitGroup) { - - server := NewServer(&ServerConfig{ - PublishTimeout: time.Millisecond * 100, - PubBuffer: 10, - Address: "0.0.0.0:8888", - }) - - // 启动定时任务 - server.StartCron() - - // 启动服务 - server.StartUp() - - <-make(chan bool) - - wg.Done() -} - -func testCron(wg *sync.WaitGroup) { - - server := NewCron(&CronConfig{ - Address: "localhost:8888", - }) - defer server.Conn.Close() - - // 创建一个cron实例 精确到秒 - c := cron.New(cron.WithSeconds()) - - // 每隔15秒执行一次 - _, _ = c.AddFunc("*/15 * * * * *", func() { - - server.Send(&pb.PublishRequest{ - Id: gouuid.GetUuId(), - Value: prefix, - Method: "wechat.1.send", - Ip: "127.0.0.1", - }) - - }) - - // 每隔30秒执行一次 - _, _ = c.AddFunc("*/30 * * * * *", func() { - - server.Send(&pb.PublishRequest{ - Id: gouuid.GetUuId(), - Value: prefix, - Method: "wechat.2.send", - Ip: "14.155.157.19", - }) - - }) - - // 启动任务 - c.Start() - - // 关闭任务 - defer c.Stop() - select {} - - wg.Done() -} - -func testWorker1(wg *sync.WaitGroup) { - - server := NewWorker(&WorkerConfig{ - Address: "localhost:8888", - ClientIp: "127.0.0.1", - }) - defer server.Conn.Close() - - // 订阅服务 - stream := server.SubscribeCron() - - // 启动任务，会想过滤器函数，订阅者应该收到的信息为 cron:任务名称 - //stream := server.StartCron() - - // 阻塞遍历流，输出结果 - for { - reply, err := stream.Recv() - if io.EOF == err { - log.Println("[跑业务1]已关闭:", err.Error()) - break - } - if nil != err { - log.Println("[跑业务1]异常:", err.Error()) - break - } - log.Printf("[跑业务1]{收到}编号：%s 方法：%s\n", reply.GetId(), reply.GetMethod()) - } - - wg.Done() -} - -func testWorker2(wg *sync.WaitGroup) { - - server := NewWorker(&WorkerConfig{ - Address: "localhost:8888", - ClientIp: "14.155.157.19", - }) - defer server.Conn.Close() - - // 订阅服务 - stream := server.SubscribeCron() - - // 启动任务，会想过滤器函数，订阅者应该收到的信息为 cron:任务名称 - //stream := server.StartCron() - - // 阻塞遍历流，输出结果 - for { - reply, err := stream.Recv() - if io.EOF == err { - log.Println("[跑业务2]已关闭:", err.Error()) - break - } - if nil != err { - log.Println("[跑业务2]异常:", err.Error()) - break - } - log.Printf("[跑业务2]{收到}编号：%s 方法：%s\n", reply.GetId(), reply.GetMethod()) - } - - wg.Done() -}