github-forks
/
redisson
mirror of https://github.com/redisson/redisson.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

refactoring

Browse Source
pull/4428/head
Nikita Koksharov 3 years ago
parent 298abd331a
commit 64805cf366
8 changed files with 79 additions and 37 deletions
Show Stats Download Patch File Download Diff File

45
redisson/src/main/java/org/redisson/RedissonExecutorService.java
Unescape Escape View File

@ -616,16 +616,24 @@ public class RedissonExecutorService implements RScheduledExecutorService {
return new RedissonExecutorBatchFuture(future, result);
}
private String generateId() {
byte[] id = new byte[16];
ThreadLocalRandom.current().nextBytes(id);
return ByteBufUtil.hexDump(id);
}
protected TaskParameters createTaskParameters(Callable<?> task) {
ClassBody classBody = getClassBody(task);
byte[] state = encode(task);
return new TaskParameters(classBody.getClazzName(), classBody.getClazz(), classBody.getLambda(), state);
String id = "00" + generateId();
return new TaskParameters(id, classBody.getClazzName(), classBody.getClazz(), classBody.getLambda(), state);
}
protected TaskParameters createTaskParameters(Runnable task) {
ClassBody classBody = getClassBody(task);
byte[] state = encode(task);
return new TaskParameters(classBody.getClazzName(), classBody.getClazz(), classBody.getLambda(), state);
String id = "00" + generateId();
return new TaskParameters(id, classBody.getClazzName(), classBody.getClazz(), classBody.getLambda(), state);
}
@Override
@ -901,10 +909,7 @@ public class RedissonExecutorService implements RScheduledExecutorService {
ClassBody classBody = getClassBody(task);
byte[] state = encode(task);
long startTime = System.currentTimeMillis() + unit.toMillis(delay);
ScheduledParameters params = new ScheduledParameters(classBody.getClazzName(), classBody.getClazz(), classBody.getLambda(), state, startTime);
if (timeToLive > 0) {
params.setTtl(ttlUnit.toMillis(timeToLive));
}
ScheduledParameters params = createScheduledParameters(timeToLive, ttlUnit, classBody, state, startTime);
RemotePromise<Void> result = (RemotePromise<Void>) asyncScheduledService.scheduleRunnable(params).toCompletableFuture();
addListener(result);
return createFuture(result, startTime);
@ -924,15 +929,21 @@ public class RedissonExecutorService implements RScheduledExecutorService {
ClassBody classBody = getClassBody(task);
byte[] state = encode(task);
long startTime = System.currentTimeMillis() + unit.toMillis(delay);
ScheduledParameters params = new ScheduledParameters(classBody.getClazzName(), classBody.getClazz(), classBody.getLambda(), state, startTime);
if (timeToLive > 0) {
params.setTtl(ttlUnit.toMillis(timeToLive));
}
ScheduledParameters params = createScheduledParameters(timeToLive, ttlUnit, classBody, state, startTime);
RemotePromise<V> result = (RemotePromise<V>) asyncScheduledService.scheduleCallable(params).toCompletableFuture();
addListener(result);
return createFuture(result, startTime);
}
private ScheduledParameters createScheduledParameters(long timeToLive, TimeUnit ttlUnit, ClassBody classBody, byte[] state, long startTime) {
String id = generateScheduledTaskId();
ScheduledParameters params = new ScheduledParameters(id, classBody.getClazzName(), classBody.getClazz(), classBody.getLambda(), state, startTime);
if (timeToLive > 0) {
params.setTtl(ttlUnit.toMillis(timeToLive));
}
return params;
}
@Override
public RScheduledFuture<?> scheduleAtFixedRate(Runnable task, long initialDelay, long period, TimeUnit unit) {
RedissonScheduledFuture<?> future = (RedissonScheduledFuture<?>) scheduleAtFixedRateAsync(task, initialDelay, period, unit);
@ -947,7 +958,9 @@ public class RedissonExecutorService implements RScheduledExecutorService {
ClassBody classBody = getClassBody(task);
byte[] state = encode(task);
long startTime = System.currentTimeMillis() + unit.toMillis(initialDelay);
ScheduledAtFixedRateParameters params = new ScheduledAtFixedRateParameters();
String id = generateScheduledTaskId();
ScheduledAtFixedRateParameters params = new ScheduledAtFixedRateParameters(id);
params.setClassName(classBody.getClazzName());
params.setClassBody(classBody.getClazz());
params.setLambdaBody(classBody.getLambda());
@ -960,6 +973,10 @@ public class RedissonExecutorService implements RScheduledExecutorService {
return createFuture(result, startTime);
}
private String generateScheduledTaskId() {
return "01" + generateId();
}
@Override
public RScheduledFuture<?> schedule(Runnable task, CronSchedule cronSchedule) {
RedissonScheduledFuture<?> future = (RedissonScheduledFuture<?>) scheduleAsync(task, cronSchedule);
@ -980,7 +997,8 @@ public class RedissonExecutorService implements RScheduledExecutorService {
}
long startTime = startDate.toInstant().toEpochMilli();
ScheduledCronExpressionParameters params = new ScheduledCronExpressionParameters();
String id = generateScheduledTaskId();
ScheduledCronExpressionParameters params = new ScheduledCronExpressionParameters(id);
params.setClassName(classBody.getClazzName());
params.setClassBody(classBody.getClazz());
params.setLambdaBody(classBody.getLambda());
@ -1015,7 +1033,8 @@ public class RedissonExecutorService implements RScheduledExecutorService {
byte[] state = encode(task);
long startTime = System.currentTimeMillis() + unit.toMillis(initialDelay);
ScheduledWithFixedDelayParameters params = new ScheduledWithFixedDelayParameters();
String id = generateScheduledTaskId();
ScheduledWithFixedDelayParameters params = new ScheduledWithFixedDelayParameters(id);
params.setClassName(classBody.getClazzName());
params.setClassBody(classBody.getClazz());
params.setLambdaBody(classBody.getLambda());

8
redisson/src/main/java/org/redisson/executor/ScheduledTasksService.java
Unescape Escape View File

@ -15,7 +15,6 @@
*/
package org.redisson.executor;
import io.netty.buffer.ByteBufUtil;
import org.redisson.RedissonExecutorService;
import org.redisson.api.RFuture;
import org.redisson.client.codec.Codec;
@ -30,7 +29,6 @@ import org.redisson.remote.ResponseEntry;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadLocalRandom;
/**
*
@ -52,7 +50,6 @@ public class ScheduledTasksService extends TasksService {
@Override
protected CompletableFuture<Boolean> addAsync(String requestQueueName, RemoteServiceRequest request) {
ScheduledParameters params = (ScheduledParameters) request.getArgs()[0];
params.setRequestId(request.getId());
long expireTime = 0;
if (params.getTtl() > 0) {
@ -143,10 +140,7 @@ public class ScheduledTasksService extends TasksService {
@Override
protected String generateRequestId(Object[] args) {
if (requestId == null) {
byte[] id = new byte[17];
ThreadLocalRandom.current().nextBytes(id);
id[0] = 01;
return ByteBufUtil.hexDump(id);
return super.generateRequestId(args);
}
return requestId;
}

13
redisson/src/main/java/org/redisson/executor/TasksService.java
Unescape Escape View File

@ -15,7 +15,6 @@
*/
package org.redisson.executor;
import io.netty.buffer.ByteBufUtil;
import org.redisson.RedissonExecutorService;
import org.redisson.api.RFuture;
import org.redisson.api.RMap;
@ -29,7 +28,10 @@ import org.redisson.misc.CompletableFutureWrapper;
import org.redisson.remote.*;
import java.util.Arrays;
import java.util.concurrent.*;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
/**
*
@ -109,7 +111,6 @@ public class TasksService extends BaseRemoteService {
protected CompletableFuture<Boolean> addAsync(String requestQueueName, RemoteServiceRequest request) {
TaskParameters params = (TaskParameters) request.getArgs()[0];
params.setRequestId(request.getId());
long retryStartTime = 0;
if (tasksRetryInterval > 0) {
@ -181,10 +182,8 @@ public class TasksService extends BaseRemoteService {
@Override
protected String generateRequestId(Object[] args) {
byte[] id = new byte[17];
ThreadLocalRandom.current().nextBytes(id);
id[0] = 00;
return ByteBufUtil.hexDump(id);
TaskParameters params = (TaskParameters) args[0];
return params.getRequestId();
}
public RFuture<Boolean> cancelExecutionAsync(String requestId) {

7
redisson/src/main/java/org/redisson/executor/params/ScheduledAtFixedRateParameters.java
Unescape Escape View File

@ -26,6 +26,13 @@ public class ScheduledAtFixedRateParameters extends ScheduledParameters {
private String executorId;
private long spentTime;
public ScheduledAtFixedRateParameters() {
}
public ScheduledAtFixedRateParameters(String requestId) {
super(requestId);
}
public long getSpentTime() {
return spentTime;
}

7
redisson/src/main/java/org/redisson/executor/params/ScheduledCronExpressionParameters.java
Unescape Escape View File

@ -26,6 +26,13 @@ public class ScheduledCronExpressionParameters extends ScheduledParameters {
private String timezone;
private String executorId;
public ScheduledCronExpressionParameters() {
}
public ScheduledCronExpressionParameters(String requestId) {
super(requestId);
}
public String getCronExpression() {
return cronExpression;
}

8
redisson/src/main/java/org/redisson/executor/params/ScheduledParameters.java
Unescape Escape View File

@ -27,8 +27,12 @@ public class ScheduledParameters extends TaskParameters {
public ScheduledParameters() {
}
public ScheduledParameters(String className, byte[] classBody, byte[] lambdaBody, byte[] state, long startTime) {
super(className, classBody, lambdaBody, state);
public ScheduledParameters(String requestId) {
super(requestId);
}
public ScheduledParameters(String id, String className, byte[] classBody, byte[] lambdaBody, byte[] state, long startTime) {
super(id, className, classBody, lambdaBody, state);
this.startTime = startTime;
}

7
redisson/src/main/java/org/redisson/executor/params/ScheduledWithFixedDelayParameters.java
Unescape Escape View File

@ -25,6 +25,13 @@ public class ScheduledWithFixedDelayParameters extends ScheduledParameters {
private long delay;
private String executorId;
public ScheduledWithFixedDelayParameters() {
}
public ScheduledWithFixedDelayParameters(String requestId) {
super(requestId);
}
public long getDelay() {
return delay;
}

7
redisson/src/main/java/org/redisson/executor/params/TaskParameters.java
Unescape Escape View File

@ -36,8 +36,13 @@ public class TaskParameters implements Serializable {
public TaskParameters() {
}
public TaskParameters(String className, byte[] classBody, byte[] lambdaBody, byte[] state) {
public TaskParameters(String requestId) {
this.requestId = requestId;
}
public TaskParameters(String requestId, String className, byte[] classBody, byte[] lambdaBody, byte[] state) {
super();
this.requestId = requestId;
this.className = className;
this.classBody = classBody;
this.state = state;

Write Preview
Loading…
Cancel
Save