- update
continuous-integration/drone/push Build is passing
Details
continuous-integration/drone/push Build is passing
Details
parent
2c4e3cd20c
commit
f95054baa8
@ -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 {}
|
||||
}
|
||||
@ -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()
|
||||
}
|
||||
@ -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()
|
||||
}
|
||||
Loading…
Reference in new issue