dtapps
/
gojobs
1
0
Fork 0
Code Issues Pull Requests Projects Releases 99 Wiki Activity

- 调整任务
continuous-integration/drone/push Build was killed Details

Browse Source
master
李光春 2 years ago
parent 63d838f853
commit 7c288242ef
6 changed files with 83 additions and 60 deletions
Show Stats Download Patch File Download Diff File

30
jobs_test.go
Unescape View File

@ -29,8 +29,10 @@ func testServer(wg *sync.WaitGroup) {
Address: "0.0.0.0:8888",
})
// 启动定时任务
server.StartCron()
// 启动服务
server.StartUp()
<-make(chan bool)
@ -52,10 +54,10 @@ func testCron(wg *sync.WaitGroup) {
_, _ = c.AddFunc("*/15 * * * * *", func() {
server.Send(&pb.PublishRequest{
Id: gouuid.GetUuId(),
//Value: prefix + "wechat.send" + " 我是定时任务",
Value: prefixSprintf("127.0.0.1") + "wechat.send" + " 我是定时任务",
Ip: "127.0.0.1",
Id: gouuid.GetUuId(),
Value: prefix,
Method: "wechat.1.send",
Ip: "127.0.0.1",
})
})
@ -64,10 +66,10 @@ func testCron(wg *sync.WaitGroup) {
_, _ = c.AddFunc("*/30 * * * * *", func() {
server.Send(&pb.PublishRequest{
Id: gouuid.GetUuId(),
//Value: prefix + "wechat.send" + " 我是定时任务",
Value: prefixSprintf("14.155.157.19") + "wechat.send" + " 我是定时任务",
Ip: "14.155.157.19",
Id: gouuid.GetUuId(),
Value: prefix,
Method: "wechat.2.send",
Ip: "14.155.157.19",
})
})
@ -91,10 +93,10 @@ func testWorker1(wg *sync.WaitGroup) {
defer server.Conn.Close()
// 订阅服务
server.SubscribeCron()
stream := server.SubscribeCron()
// 启动任务，会想过滤器函数，订阅者应该收到的信息为 cron:任务名称
stream := server.StartCron()
//stream := server.StartCron()
// 阻塞遍历流，输出结果
for {
@ -107,7 +109,7 @@ func testWorker1(wg *sync.WaitGroup) {
log.Println("[跑业务1]异常:", err.Error())
break
}
log.Println("[跑业务1]收到:", reply)
log.Printf("[跑业务1]{收到}编号：%s 方法：%s\n", reply.GetId(), reply.GetMethod())
}
wg.Done()
@ -122,10 +124,10 @@ func testWorker2(wg *sync.WaitGroup) {
defer server.Conn.Close()
// 订阅服务
server.SubscribeCron()
stream := server.SubscribeCron()
// 启动任务，会想过滤器函数，订阅者应该收到的信息为 cron:任务名称
stream := server.StartCron()
//stream := server.StartCron()
// 阻塞遍历流，输出结果
for {
@ -138,7 +140,7 @@ func testWorker2(wg *sync.WaitGroup) {
log.Println("[跑业务2]异常:", err.Error())
break
}
log.Println("[跑业务2]收到:", reply)
log.Printf("[跑业务2]{收到}编号：%s 方法：%s\n", reply.GetId(), reply.GetMethod())
}
wg.Done()

89
pb/pubsub.pb.go
Unescape View File

@ -30,9 +30,10 @@ type PublishRequest struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
Ip string `protobuf:"bytes,3,opt,name=ip,proto3" json:"ip,omitempty"`
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
Method string `protobuf:"bytes,3,opt,name=method,proto3" json:"method,omitempty"`
Ip string `protobuf:"bytes,4,opt,name=ip,proto3" json:"ip,omitempty"`
}
func (x *PublishRequest) Reset() {
@ -81,6 +82,13 @@ func (x *PublishRequest) GetValue() string {
return ""
}
func (x *PublishRequest) GetMethod() string {
if x != nil {
return x.Method
}
return ""
}
func (x *PublishRequest) GetIp() string {
if x != nil {
return x.Ip
@ -158,9 +166,10 @@ type SubscribeRequest struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
Ip string `protobuf:"bytes,3,opt,name=ip,proto3" json:"ip,omitempty"`
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
Method string `protobuf:"bytes,3,opt,name=method,proto3" json:"method,omitempty"`
Ip string `protobuf:"bytes,4,opt,name=ip,proto3" json:"ip,omitempty"`
}
func (x *SubscribeRequest) Reset() {
@ -209,6 +218,13 @@ func (x *SubscribeRequest) GetValue() string {
return ""
}
func (x *SubscribeRequest) GetMethod() string {
if x != nil {
return x.Method
}
return ""
}
func (x *SubscribeRequest) GetIp() string {
if x != nil {
return x.Ip
@ -222,9 +238,9 @@ type SubscribeResponse struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
Ip string `protobuf:"bytes,3,opt,name=ip,proto3" json:"ip,omitempty"`
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
Method string `protobuf:"bytes,3,opt,name=method,proto3" json:"method,omitempty"`
}
func (x *SubscribeResponse) Reset() {
@ -273,9 +289,9 @@ func (x *SubscribeResponse) GetValue() string {
return ""
}
func (x *SubscribeResponse) GetIp() string {
func (x *SubscribeResponse) GetMethod() string {
if x != nil {
return x.Ip
return x.Method
}
return ""
}
@ -284,34 +300,37 @@ var File_pb_pubsub_proto protoreflect.FileDescriptor
var file_pb_pubsub_proto_rawDesc = []byte{
0x0a, 0x0f, 0x70, 0x62, 0x2f, 0x70, 0x75, 0x62, 0x73, 0x75, 0x62, 0x2e, 0x70, 0x72, 0x6f, 0x74,
0x6f, 0x12, 0x02, 0x70, 0x62, 0x22, 0x46, 0x0a, 0x0e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68,
0x6f, 0x12, 0x02, 0x70, 0x62, 0x22, 0x5e, 0x0a, 0x0e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68,
0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20,
0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65,
0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x0e, 0x0a,
0x02, 0x69, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x70, 0x22, 0x47, 0x0a,
0x0f, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64,
0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x03, 0x20, 0x01,
0x28, 0x09, 0x52, 0x02, 0x69, 0x70, 0x22, 0x48, 0x0a, 0x10, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72,
0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64,
0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61,
0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65,
0x12, 0x0e, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x70,
0x22, 0x49, 0x0a, 0x11, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x73,
0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x16, 0x0a,
0x06, 0x6d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x6d,
0x65, 0x74, 0x68, 0x6f, 0x64, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x04, 0x20, 0x01, 0x28,
0x09, 0x52, 0x02, 0x69, 0x70, 0x22, 0x47, 0x0a, 0x0f, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68,
0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01,
0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75,
0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x0e,
0x0a, 0x02, 0x69, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x70, 0x22, 0x60,
0x0a, 0x10, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02,
0x69, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x6d, 0x65, 0x74, 0x68,
0x6f, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x6d, 0x65, 0x74, 0x68, 0x6f, 0x64,
0x12, 0x0e, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x70,
0x22, 0x51, 0x0a, 0x11, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x73,
0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02,
0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x0e, 0x0a, 0x02, 0x69,
0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x70, 0x32, 0x78, 0x0a, 0x06, 0x50,
0x75, 0x62, 0x53, 0x75, 0x62, 0x12, 0x32, 0x0a, 0x07, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68,
0x12, 0x12, 0x2e, 0x70, 0x62, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x71,
0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, 0x70, 0x62, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73,
0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x3a, 0x0a, 0x09, 0x53, 0x75, 0x62,
0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x12, 0x14, 0x2e, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73,
0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x15, 0x2e, 0x70,
0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f,
0x6e, 0x73, 0x65, 0x30, 0x01, 0x42, 0x07, 0x5a, 0x05, 0x2e, 0x2e, 0x2f, 0x70, 0x62, 0x62, 0x06,
0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x6d,
0x65, 0x74, 0x68, 0x6f, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x6d, 0x65, 0x74,
0x68, 0x6f, 0x64, 0x32, 0x78, 0x0a, 0x06, 0x50, 0x75, 0x62, 0x53, 0x75, 0x62, 0x12, 0x32, 0x0a,
0x07, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x12, 0x12, 0x2e, 0x70, 0x62, 0x2e, 0x50, 0x75,
0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, 0x70,
0x62, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
0x65, 0x12, 0x3a, 0x0a, 0x09, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x12, 0x14,
0x2e, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71,
0x75, 0x65, 0x73, 0x74, 0x1a, 0x15, 0x2e, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72,
0x69, 0x62, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x30, 0x01, 0x42, 0x07, 0x5a,
0x05, 0x2e, 0x2e, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (

8
pb/pubsub.proto
Unescape View File

@ -19,7 +19,8 @@ service PubSub {
message PublishRequest {
string id = 1;
string value = 2;
string ip = 3;
string method = 3;
string ip = 4;
}
// 响应消息
@ -33,12 +34,13 @@ message PublishResponse {
message SubscribeRequest {
string id = 1;
string value = 2;
string ip = 3;
string method = 3;
string ip = 4;
}
// 响应消息
message SubscribeResponse {
string id = 1;
string value = 2;
string ip = 3;
string method = 3;
}

9
pb/pubsub.server.go
Unescape View File

@ -41,12 +41,9 @@ func (p *PubSubServerService) Subscribe(req *SubscribeRequest, stream PubSub_Sub
// func(v interface{}) 定义函数过滤的规则
// SubscribeTopic 返回一个chan interface{}
log.Printf("[服务中转]{订阅}编号：%s 类型：%s ip地址：%s\n", req.GetId(), req.GetValue(), req.GetIp())
ch := p.pub.SubscribeTopic(func(v interface{}) bool {
log.Printf("[服务中转]{订阅}主题：%v\n", v)
log.Printf("[服务中转]{订阅}主题：%+v\n", v)
// 接收数据是string，并且key是以arg为前缀的
if key, ok := v.(string); ok {
@ -57,6 +54,7 @@ func (p *PubSubServerService) Subscribe(req *SubscribeRequest, stream PubSub_Sub
return false
})
log.Printf("[服务中转]{订阅}编号：%s 类型：%s 方法：%s ip地址：%s\n", req.GetId(), req.GetValue(), req.GetMethod(), req.GetIp())
log.Println("[服务中转]{订阅}工作线：", ch)
log.Println("[服务中转]{订阅}当前工作线数量：", p.pub.Len())
@ -65,8 +63,9 @@ func (p *PubSubServerService) Subscribe(req *SubscribeRequest, stream PubSub_Sub
log.Println("[服务中转]{订阅}for ch：", ch)
log.Println("[服务中转]{订阅}for v：", v)
err := stream.Send(&SubscribeResponse{
Value: v.(string),
Ip: "",
Id: req.GetId(),
Value: req.GetValue(),
Method: req.GetMethod(),
})
if err != nil {
log.Println("[服务中转]{订阅}任务分配失败 ", err.Error())

2
server.go
Unescape View File

@ -65,7 +65,7 @@ func (s *Server) StartCron() {
}()
}
// StartUp 启动
// StartUp 启动服务
func (s *Server) StartUp() {
// 监听本地端口

5
worker.go
Unescape View File

@ -50,8 +50,8 @@ func NewWorker(config *WorkerConfig) *Worker {
}
// SubscribeCron 订阅服务
func (w *Worker) SubscribeCron() {
_, err := w.Pub.Subscribe(context.Background(), &pb.SubscribeRequest{
func (w *Worker) SubscribeCron() pb.PubSub_SubscribeClient {
stream, err := w.Pub.Subscribe(context.Background(), &pb.SubscribeRequest{
Id: gouuid.GetUuId(),
Value: prefix,
Ip: w.ClientIp,
@ -59,6 +59,7 @@ func (w *Worker) SubscribeCron() {
if err != nil {
panic("[工作线]{订阅服务失败}" + err.Error())
}
return stream
}
// StartCron 启动任务

Write Preview
Loading…
Cancel
Save