-
Notifications
You must be signed in to change notification settings - Fork 95
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #139 from stelin/main
Update status
- Loading branch information
Showing
28 changed files
with
278 additions
and
31 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
49 changes: 49 additions & 0 deletions
49
...r-cluster/src/main/java/io/openjob/server/cluster/executor/WorkerDelayStatusExecutor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| package io.openjob.server.cluster.executor; | ||
|
|
||
| import io.openjob.common.request.WorkerDelayStatusRequest; | ||
| import io.openjob.common.task.TaskQueue; | ||
| import io.openjob.server.cluster.task.WorkerDelayStatusConsumer; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| /** | ||
| * @author stelin [email protected] | ||
| * @since 1.0.6 | ||
| */ | ||
| @Slf4j | ||
| @Component | ||
| public class WorkerDelayStatusExecutor { | ||
| private final TaskQueue<WorkerDelayStatusRequest> queue; | ||
|
|
||
| /** | ||
| * New | ||
| */ | ||
| public WorkerDelayStatusExecutor() { | ||
| this.queue = new TaskQueue<>(0L, 1024); | ||
|
|
||
| //Consumer | ||
| WorkerDelayStatusConsumer consumer = new WorkerDelayStatusConsumer( | ||
| 0L, | ||
| 1, | ||
| 16, | ||
| "Openjob-heartbeat-executor", | ||
| 50, | ||
| "Openjob-heartbeat-consumer", | ||
| this.queue | ||
| ); | ||
| consumer.start(); | ||
| } | ||
|
|
||
| /** | ||
| * Submit request | ||
| * | ||
| * @param request request | ||
| */ | ||
| public void submit(WorkerDelayStatusRequest request) { | ||
| try { | ||
| this.queue.submit(request); | ||
| } catch (InterruptedException e) { | ||
| log.error("Worker heartbeat submit failed!", e); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
49 changes: 49 additions & 0 deletions
49
...r-cluster/src/main/java/io/openjob/server/cluster/executor/WorkerJobInstanceExecutor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| package io.openjob.server.cluster.executor; | ||
|
|
||
| import io.openjob.common.request.WorkerJobInstanceStatusRequest; | ||
| import io.openjob.common.task.TaskQueue; | ||
| import io.openjob.server.cluster.task.WorkerJobInstanceConsumer; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| /** | ||
| * @author stelin [email protected] | ||
| * @since 1.0.6 | ||
| */ | ||
| @Slf4j | ||
| @Component | ||
| public class WorkerJobInstanceExecutor { | ||
| private final TaskQueue<WorkerJobInstanceStatusRequest> queue; | ||
|
|
||
| /** | ||
| * New | ||
| */ | ||
| public WorkerJobInstanceExecutor() { | ||
| this.queue = new TaskQueue<>(0L, 1024); | ||
|
|
||
| //Consumer | ||
| WorkerJobInstanceConsumer consumer = new WorkerJobInstanceConsumer( | ||
| 0L, | ||
| 1, | ||
| 32, | ||
| "Openjob-heartbeat-executor", | ||
| 50, | ||
| "Openjob-heartbeat-consumer", | ||
| this.queue | ||
| ); | ||
| consumer.start(); | ||
| } | ||
|
|
||
| /** | ||
| * Submit request | ||
| * | ||
| * @param request request | ||
| */ | ||
| public void submit(WorkerJobInstanceStatusRequest request) { | ||
| try { | ||
| this.queue.submit(request); | ||
| } catch (InterruptedException e) { | ||
| log.error("Worker heartbeat submit failed!", e); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
30 changes: 30 additions & 0 deletions
30
...ob-server-cluster/src/main/java/io/openjob/server/cluster/service/WorkerDelayService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| package io.openjob.server.cluster.service; | ||
|
|
||
| import io.openjob.common.request.WorkerDelayStatusRequest; | ||
| import io.openjob.server.cluster.executor.WorkerDelayStatusExecutor; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| /** | ||
| * @author stelin [email protected] | ||
| * @since 1.0.6 | ||
| */ | ||
| @Service | ||
| public class WorkerDelayService { | ||
|
|
||
| private final WorkerDelayStatusExecutor workerDelayStatusExecutor; | ||
|
|
||
| @Autowired | ||
| public WorkerDelayService(WorkerDelayStatusExecutor workerDelayStatusExecutor) { | ||
| this.workerDelayStatusExecutor = workerDelayStatusExecutor; | ||
| } | ||
|
|
||
| /** | ||
| * Handle delay status. | ||
| * | ||
| * @param workerDelayStatusRequest workerDelayStatusRequest | ||
| */ | ||
| public void handleDelayStatus(WorkerDelayStatusRequest workerDelayStatusRequest) { | ||
| this.workerDelayStatusExecutor.submit(workerDelayStatusRequest); | ||
| } | ||
| } |
49 changes: 49 additions & 0 deletions
49
...erver-cluster/src/main/java/io/openjob/server/cluster/task/WorkerDelayStatusConsumer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| package io.openjob.server.cluster.task; | ||
|
|
||
| import io.openjob.common.OpenjobSpringContext; | ||
| import io.openjob.common.request.WorkerDelayStatusRequest; | ||
| import io.openjob.common.task.BaseConsumer; | ||
| import io.openjob.common.task.TaskQueue; | ||
| import io.openjob.server.scheduler.service.DelayInstanceService; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| /** | ||
| * @author stelin [email protected] | ||
| * @since 1.0.6 | ||
| */ | ||
| @Slf4j | ||
| public class WorkerDelayStatusConsumer extends BaseConsumer<WorkerDelayStatusRequest> { | ||
| public WorkerDelayStatusConsumer(Long id, | ||
| Integer consumerCoreThreadNum, | ||
| Integer consumerMaxThreadNum, | ||
| String consumerThreadName, | ||
| Integer pollSize, | ||
| String pollThreadName, | ||
| TaskQueue<WorkerDelayStatusRequest> queues) { | ||
| super(id, consumerCoreThreadNum, consumerMaxThreadNum, consumerThreadName, pollSize, pollThreadName, queues, 2000L, 1000L); | ||
| } | ||
|
|
||
| @Override | ||
| public void consume(Long id, List<WorkerDelayStatusRequest> tasks) { | ||
| this.consumerExecutor.submit(new WorkerDelayStatusRunnable(tasks)); | ||
| } | ||
|
|
||
| private static class WorkerDelayStatusRunnable implements Runnable { | ||
| private final List<WorkerDelayStatusRequest> tasks; | ||
|
|
||
| private WorkerDelayStatusRunnable(List<WorkerDelayStatusRequest> tasks) { | ||
| this.tasks = tasks; | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| try { | ||
| this.tasks.forEach(r -> OpenjobSpringContext.getBean(DelayInstanceService.class).handleConsumerDelayStatus(r)); | ||
| } catch (Throwable throwable) { | ||
| log.error("Worker delay status consume failed!", throwable); | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,13 +5,15 @@ | |
| import io.openjob.common.task.BaseConsumer; | ||
| import io.openjob.common.task.TaskQueue; | ||
| import io.openjob.server.cluster.service.WorkerHeartbeatService; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| /** | ||
| * @author stelin [email protected] | ||
| * @since 1.0.3 | ||
| */ | ||
| @Slf4j | ||
| public class WorkerHeartConsumer extends BaseConsumer<WorkerHeartbeatRequest> { | ||
|
|
||
| public WorkerHeartConsumer(Long id, | ||
|
|
@@ -41,7 +43,11 @@ private WorkerHeartbeatConsumerRunnable(List<WorkerHeartbeatRequest> tasks) { | |
|
|
||
| @Override | ||
| public void run() { | ||
| OpenjobSpringContext.getBean(WorkerHeartbeatService.class).batchHeartbeat(this.tasks); | ||
| try { | ||
| OpenjobSpringContext.getBean(WorkerHeartbeatService.class).batchHeartbeat(this.tasks); | ||
| } catch (Throwable throwable) { | ||
| log.error("Worker heartbeat failed!", throwable); | ||
| } | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.