新增:1.redtimer 代码包暂时合并到此工程

2.新增Timers.tryDelay、Timers.delay 方法
This commit is contained in:
2021-11-03 19:56:27 +08:00
parent 257e840e7e
commit 89d474284c
16 changed files with 3772 additions and 14 deletions

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,262 @@
package com.zdemo.cache_;
import com.zdemo.cachex.MyRedisCacheSource;
import org.redkale.convert.json.JsonFactory;
import org.redkale.net.AsyncIOGroup;
import org.redkale.util.AnyValue;
import org.redkale.util.ResourceFactory;
import java.util.Map;
import static org.redkale.boot.Application.RESNAME_APP_GROUP;
public class RedisTest {
static MyRedisCacheSource<String> source;
static MyRedisCacheSource<Integer> sourceInt;
static {
AnyValue.DefaultAnyValue conf = new AnyValue.DefaultAnyValue().addValue("maxconns", "10");
conf.addValue("node", new AnyValue.DefaultAnyValue().addValue("addr", "47.111.150.118").addValue("port", "6064").addValue("password", "*Zhong9307!").addValue("db", 1));
final AsyncIOGroup asyncGroup = new AsyncIOGroup(8192, 16);
asyncGroup.start();
ResourceFactory.root().register(RESNAME_APP_GROUP, asyncGroup);
source = new MyRedisCacheSource();
ResourceFactory.root().inject(source);
source.init(null);
source.defaultConvert = JsonFactory.root().getConvert();
source.init(conf);
// int
sourceInt = new MyRedisCacheSource<Integer>();
ResourceFactory.root().inject(sourceInt);
sourceInt.init(null);
sourceInt.defaultConvert = JsonFactory.root().getConvert();
sourceInt.init(conf);
sourceInt.initValueType(Integer.class);
}
public static void main(String[] args) {
//System.out.println(source.remove("a", "b"));
// bit
/*source.initValueType(Integer.class);
source.remove("a");
boolean a = source.getBit("a", 1);
System.out.println(a);
source.setBit("a", 1, true);
a = source.getBit("a", 1);
System.out.println("bit-a-1: " + a);
source.setBit("a", 1, false);
a = source.getBit("a", 1);
System.out.println("bit-a-1: " + a);*/
/*source.remove("a");
// setnx
System.out.println(source.setnx("a", 1));
source.remove("a");
System.out.println(source.setnx("a", 1));
// set
source.remove("abx1");
source.appendSetItems("abx1", "a", "b", "c");
List<String> list = source.srandomItems("abx1", 2);
String str = source.srandomItem("abx1"); //r
System.out.println(list);//[r1, r2] */
/*int[] arr = {0};
ExecutorService executor = Executors.newFixedThreadPool(10);
CountDownLatch latch = new CountDownLatch(1000);
for (int i = 0; i < 1000; i++) {
executor.submit(() -> {
try {
source.lock("c", 1000);
arr[0]++;
// System.out.println("Thread: " + Thread.currentThread().getName());
// Thread.sleep(10);
} catch (Exception e) {
e.printStackTrace();
} finally {
source.unlock("c");
latch.countDown();
}
});
}
try {
latch.await();
System.out.println("n=" + arr[0]);
executor.shutdown();
} catch (InterruptedException e) {
e.printStackTrace();
}*/
/*List<String> list = (List) source.getCollection("gamerank-comment-stat");
System.out.println(list);*/
/*for (int i = 0; i < 10; i++) {
String brpop = source.brpop("z", 2);
System.out.println(brpop);
}*/
// key 测试
/*source.set("a", "123321");
System.out.println(source.get("a")); // 123321
System.out.println(source.getTtl("a")); // -1
System.out.println(source.getPttl("a")); // -1
System.out.println(source.getPttl("x")); // -2*/
// hashmap 测试
/*source.remove("sk");
source.setHm("sk", "a", "1");
source.setHm("sk", "b", "2");
System.out.println(source.getHm("sk", "a")); // 1
source.remove("sk");
source.setHms("sk", Map.of("b", "5", "c", "3", "a", "1"));
source.hdel("sk", "a");
Map map = source.getHms("sk", "a", "x", "b", "c", "f"); // {b=5, c=3}
System.out.println(map);
System.out.println(source.getHmall("sk")); //{b=5, c=3}
System.out.println(source.incrHm("sk", "b", 1.1d)); // b = 6.1
System.out.println(source.incrHm("sk", "c", 1)); // c = 4
System.out.println(source.getHmall("sk")); //{b=6.1, c=4}
System.out.println("--------------");
System.out.println(source.hexists("sk", "b")); // true
System.out.println(source.getCollectionSize("sk")); // 2*/
Map<String, String> hms = source.getHms("supportusers", "5-kfeu0f", "xxxx", "3-0kbt7u8t", "95q- ");
hms.forEach((k, v) -> {
System.out.println(k + " : " + v);
});
/*MyRedisCacheSource<String> source2 = new MyRedisCacheSource();
source2.defaultConvert = JsonFactory.root().getConvert();
source2.initValueType(String.class); //value用String类型
source2.init(conf);*/
/*Map<String, String> gcMap = source.getHmall("hot-gamecomment");
gcMap.forEach((k,v) -> {
System.out.println(k + " : " + v);
});*/
//Map<String, String> gameinfo = source.getHms("gameinfo", "22759", "22838", "10097", "22751", "22632", "22711", "22195", "15821", "10099", "16313", "11345", "10534", "22768", "22647", "22924", "18461", "15871", "17099", "22640", "22644", "10744", "10264", "18032", "22815", "13584", "10031", "22818", "22452", "22810", "10513", "10557", "15848", "11923", "15920", "22808", "20073", "22809", "15840", "12332", "15803", "10597", "22624", "17113", "19578", "22664", "22621", "20722", "16226", "10523", "12304", "10597","11923","10031");
//Map<String, String> gameinfo = source.getHms("gameinfo", "22759","22838","10097","22751","22632","22711","22195","15821","10099","16313","11345","10534","22768","22647","22924","18461","15871","17099","22363","22640","22644","10744","10264","18032","22815","13584","22818","22452","22810","10513","10557","15848","15920","22808","20073","22809","15840","12332","15803","10597","22624","17113","19578","22627","22664","22621","20722","16226","10523","12304");
/*gameinfo.forEach((k,v ) -> {
System.out.println(v);
});*/
/*source.queryKeysStartsWith("articlebean:").forEach(x -> {
System.out.println(x);
//source.remove(x);
//System.out.println(source.getHmall(x));
});*/
// list 测试
/*sourceInt.remove("list");
Collection<Integer> list = sourceInt.getCollection("list");
System.out.println(list);
for (int i = 1; i <= 10; i++) {
sourceInt.appendListItem("list", i);
}
System.out.println(sourceInt.getCollection("list")); // [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
sourceInt.appendListItems("list", 11, 12, 13);
System.out.println(sourceInt.getCollection("list")); // [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]
System.out.println(sourceInt.getCollection("list", 0, 5)); // [1, 2, 3, 4, 5]
System.out.println(sourceInt.getCollectionSize("list")); // 13
List<Integer> ids = new ArrayList<>(100);
for (int i = 0; i < 2000; i++) {
ids.add(i);
}
sourceInt.remove("abx");
sourceInt.appendListItems("abx", ids.toArray(Integer[]::new));
System.out.println(sourceInt.getCollection("abx"));
*/
/*System.out.println(sourceInt.getCollectionSize("recommend-user-quality"));
Collection<Integer> uid = sourceInt.getCollection("recommend-user-quality");
System.out.println(uid);*/
// zset 测试
/*source.initValueType(String.class); //value用Integer类型
source.remove("zx");
source.zadd("zx", Map.of("a", 1, "b", 2));
source.zadd("zx", Map.of("a", 1, "c", 5L));
source.zadd("zx", Map.of("x", 20, "j", 3.5));
source.zadd("zx", Map.of("f", System.currentTimeMillis(), "c", 5L));
source.zadd("zx", Map.of("a", 1, "c", 5L));
System.out.println(source.zincr("zx", "a", 1.34)); // 2.34
System.out.println(source.getZsetDoubleScore("zx")); // {f=1592924555704, x=20, c=5, j=3, b=2.34, a=1}
source.zrem("zx", "b", "c", "e", "x");
System.out.println(source.getZsetLongScore("zx")); // {f=1592924555704, j=3, a=2}
System.out.println("--------------");
System.out.println(source.getZsetLongScore("zx", "f"));
System.out.println(source.getZrevrank("zx", "f")); // 0
System.out.println(source.getZrank("zx", "f")); // 2
System.out.println(source.getZrank("zx", "Y")); // -1
System.out.println(source.getCollectionSize("zx")); // 3
System.out.println(source.getZset("zx"));
System.out.println(source.zexists("zx", "f", "x", "a"));*/
/*LocalDate date = LocalDate.of(2019, 12, 31);
for (int i = 0; i < 60; i++) {
LocalDate localDate = date.plusDays(-i);
String day = localDate.format(DateTimeFormatter.ISO_LOCAL_DATE);
System.out.println(String.format("mkdir %s; mv *%s*.zip %s", day, day, day));
}*/
/*MyRedisCacheSource<UserDetail> source = new MyRedisCacheSource();
source.defaultConvert = JsonFactory.root().getConvert();
source.initValueType(UserDetail.class);
source.init(conf);
Map<String, UserDetail> map = source.getHmall("user-detail");
Integer[] array = map.values().stream().map(x -> x.getUserid()).toArray(Integer[]::new);
System.out.println(JsonConvert.root().convertTo(array));
Map<Integer, UserDetail> hms = source.getHms("user-detail", 11746, 11988, 11504, 11987, 11745, 11503, 11748, 11506, 11747, 11989, 11505, 11508, 11507, 11509, 11980, 11740, 11982, 11981, 11984, 11742, 11500, 11983, 11741, 11502, 11744, 11986, 11985, 11501, 11743, 11999, 11757, 11515, 1, 11514, 11998, 11756, 2, 11517, 11516, 11758, 3, 11519, 4, 5, 11518, 6, 7, 11991, 8, 11990, 9, 11993, 11751, 11750, 11992, 11753, 11511, 11995, 11994, 11510, 11752, 11755, 11513, 11997, 11512, 11996, 11754, 11724, 11966, 11965, 11723, 11968, 11726, 11967, 11725, 11728, 11969, 11727, 11729, 11960, 11720, 11962, 11961, 11722, 11964, 11721);
System.out.println(hms.size());*/
/*source.getCollection("article-comment-list", 19, 1).forEach(x -> System.out.println(x));
while (true) {
System.out.println("---" + Utility.now() + "---");
source.getHmall("ck").forEach((k, v) -> {
System.out.println(k + ":" + v);
});
try {
Thread.sleep(60 * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}*/
}
}

View File

@@ -1,72 +0,0 @@
package com.zdemo.zhub;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.logging.Logger;
public class Delays implements Delayed, Runnable {
public Logger logger = Logger.getLogger(Delays.class.getSimpleName());
private long time; // 执行时间
private Runnable runnable; // 任务到时间执行 runnable
public Delays(long timeout, Runnable runnable) {
this.time = System.currentTimeMillis() + timeout;
this.runnable = runnable;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(time - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed other) {
if (other == this) { // compare zero ONLY if same object
return 0;
}
if (other instanceof Delays) {
Delays x = (Delays) other;
long diff = time - x.time;
if (diff < 0) {
return -1;
} else if (diff > 0) {
return 1;
}
}
long d = (getDelay(TimeUnit.NANOSECONDS) -
other.getDelay(TimeUnit.NANOSECONDS));
return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
}
@Override
public void run() {
runnable.run();
}
// ===========
public static DelayQueue<Delays> delayQueue = new DelayQueue<>();
public static void addDelay(long timeout, Runnable runnable) {
delayQueue.add(new Delays(timeout, runnable));
}
public static void tryDelay(Supplier<Boolean> supplier, long delayMillis, int maxCount) {
}
static {
new Thread(() -> {
try {
while (true) {
Delays delay = delayQueue.take();
delay.run(); //异常会导致延时队列失败
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}

View File

@@ -4,6 +4,7 @@ import com.zdemo.AbstractConsumer;
import com.zdemo.Event;
import com.zdemo.IConsumer;
import com.zdemo.IProducer;
import net.tccn.timer.Timers;
import org.redkale.service.Service;
import org.redkale.util.*;
@@ -524,14 +525,19 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
timeout = 1000 * 15;
}
// call timeout default: 15s
Delays.addDelay(timeout, () -> {
RpcResult rpcResult = rpc.buildResp(505, "请求超时");
rpc.setRpcResult(rpcResult);
Timers.delay(() -> {
synchronized (rpc) {
Rpc rpc1 = rpcMap.get(ruk);
if (rpc1 == null) {
return;
}
RpcResult rpcResult = rpc.buildResp(505, "请求超时");
rpc.setRpcResult(rpcResult);
logger.warning("rpc timeout: " + convert.convertTo(rpc));
rpc.notify();
}
});
}, timeout);
rpc.wait();
rpcMap.remove(ruk);
@@ -568,7 +574,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
}.getType(), value);
String ruk = resp.getRuk();
Rpc rpc = rpcMap.get(ruk);
Rpc rpc = rpcMap.remove(ruk);
if (rpc == null) {
return;
}

View File

@@ -0,0 +1,66 @@
package net.tccn.timer;
import net.tccn.timer.queue.TimerQueue;
import net.tccn.timer.task.Task;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;
/**
* @author: liangxianyou
*/
public class TimerExecutor {
private Logger logger = Logger.getLogger(this.getClass().getSimpleName());
private TimerQueue queue = new TimerQueue();
private ExecutorService executor;
public TimerExecutor(int n) {
executor = Executors.newFixedThreadPool(n);
start();
}
public void add(Task... task) {
for (Task t : task) {
t.setTimerExecutor(this);
queue.push(t);
logger.finest("add new task : " + t.getName());
}
}
protected void add(Task task, boolean upTime) {
task.setTimerExecutor(this);
if (upTime) task.nextTime();
queue.push(task);
}
public Task remove(String name) {
return queue.remove(name);
}
public Task get(String name) {
return queue.get(name);
}
public void start() {
new Thread(() -> {
while (true) {
try {
Task take = null;
try {
take = queue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
//执行调度
executor.execute(take);
//add(take, true); //继续添加任务到 队列
} catch (Exception e) {
e.printStackTrace();
}
}
}, "Thread-Redtimer-0").start();
}
}

View File

@@ -0,0 +1,108 @@
package net.tccn.timer;
import net.tccn.timer.scheduled.Scheduled;
import net.tccn.timer.task.Job;
import net.tccn.timer.task.Task;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;
/**
* Created by liangxianyou at 2018/7/23 14:33.
*/
public class TimerTask implements Task {
private Logger logger = Logger.getLogger(this.getClass().getSimpleName());
private long startTime = System.currentTimeMillis();
private AtomicInteger execCount = new AtomicInteger();
protected String name;
private long theTime;
private Scheduled scheduled;
private boolean isComplete;
private TimerExecutor timerExecutor;
private Job job;
public static Task by(String name, Scheduled scheduled, Job job) {
TimerTask task = new TimerTask();
task.name = name;
task.scheduled = scheduled;
task.job = job;
return task;
}
@Override
public String getName() {
return name;
}
@Override
public void setScheduled(Scheduled scheduled) {
this.scheduled = scheduled;
//this.theTime = Date.from(scheduled.theTime().atZone(ZoneId.systemDefault()).toInstant()).getTime();
}
@Override
public long nextTime() {
LocalDateTime next = scheduled.nextTime();
this.theTime = Date.from(next.atZone(ZoneId.systemDefault()).toInstant()).getTime();
/*SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
System.out.println("下次执行:"+ sdf.format(next.toInstant(ZoneOffset.of("+8")).toEpochMilli()));*/
return theTime;
}
@Override
public long theTime() {
LocalDateTime next = scheduled.theTime();
this.theTime = Date.from(next.atZone(ZoneId.systemDefault()).toInstant()).getTime();
return theTime;
}
@Override
public boolean isComplete() {
return isComplete;
}
public void setComplete(boolean complete) {
if (isComplete = complete)
timerExecutor.remove(name);
}
public int getExecCount() {
return execCount.get();
}
public TimerExecutor getTimerExecutor() {
return timerExecutor;
}
public void setTimerExecutor(TimerExecutor timerExecutor) {
this.timerExecutor = timerExecutor;
}
public long startTime() {
return startTime;
}
@Override
public void run() {
//没有完成任务,继续执行
if (!isComplete) {
int count = execCount.incrementAndGet(); // 执行次数+1
long start = System.currentTimeMillis();
job.execute(this);
long end = System.currentTimeMillis();
logger.finest(String.format("task [%s] : not complete -> %s, time: %s ms, exec count: %s.", getName(), isComplete ? "had complete" : "not complete", end - start, count));
if (!isComplete) {
timerExecutor.add(this, true);
}
}
}
}

View File

@@ -0,0 +1,45 @@
package net.tccn.timer;
import net.tccn.timer.scheduled.ScheduledCycle;
import org.redkale.util.Utility;
import java.util.function.Supplier;
public class Timers {
private static TimerExecutor timerExecutor = new TimerExecutor(1);
/**
* 本地延时重试
* @param supplier
* @param millis
* @param maxCount
*/
public static void tryDelay(Supplier<Boolean> supplier, long millis, int maxCount) {
timerExecutor.add(TimerTask.by("try-delay-task-" + Utility.uuid(), ScheduledCycle.of(0), task -> {
if (supplier.get() || task.getExecCount() == maxCount) {
task.setComplete(true);
}
if (task.getExecCount() == 1) {
task.setScheduled(ScheduledCycle.of(millis));
}
}));
}
/**
* 本地延时:延时时间极短的场景下使用 1分钟内
* @param runnable
* @param millis
*/
public static void delay(Runnable runnable, long millis) {
timerExecutor.add(TimerTask.by("delay-task-" + Utility.uuid(), ScheduledCycle.of(millis), task -> {
runnable.run();
task.setComplete(true);
}));
}
}

View File

@@ -0,0 +1,105 @@
package net.tccn.timer.queue;
import net.tccn.timer.task.Task;
import java.util.LinkedList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* Created by liangxianyou at 2018/7/23 14:07.
*/
public class TimerQueue {
private ReentrantLock lock = new ReentrantLock();
private Condition isEmpty = lock.newCondition();
private LinkedList<Task> queue = new LinkedList();
/**
* 新加调度任务
*
* @param task
*/
public void push(Task task) {
try {
lock.lock();
remove(task.getName());
int inx = queue.size();//目标坐标
while (inx > 0 && queue.get(inx - 1).theTime() > task.theTime()) {
inx--;
}
queue.add(inx, task);
isEmpty.signal();
} finally {
lock.unlock();
}
}
/**
* 调度等待执行的任务
*
* @return
* @throws InterruptedException
*/
public Task take() throws InterruptedException {
try {
lock.lock();
while (queue.size() == 0) {
isEmpty.await();
}
long currentTime = System.currentTimeMillis();
long nextTime = queue.getFirst().theTime();
if (currentTime >= nextTime) {
return queue.removeFirst();
} else {
isEmpty.await(nextTime - currentTime, TimeUnit.MILLISECONDS);
return take();
}
} finally {
lock.unlock();
}
}
/**
* 删除指定名称的任务
*
* @param name
* @return
*/
public Task remove(String name) {
return get(name, true);
}
/**
* 返回指定名称的任务
*
* @param name
* @return
*/
public Task get(String name) {
return get(name, false);
}
private Task get(String name, boolean remove) {
try {
lock.lock();
Task take = null;
for (int i = 0; i < queue.size(); i++) {
if (name.equals(queue.get(i).getName())) {
take = queue.get(i);
}
}
if (remove && take != null) {
queue.remove(take);
}
isEmpty.signal();
return take;
} finally {
lock.unlock();
}
}
}

View File

@@ -0,0 +1,23 @@
package net.tccn.timer.scheduled;
import java.time.LocalDateTime;
/**
* @author: liangxianyou at 2018/8/5 17:35.
*/
public interface Scheduled {
/**
* 下次执行时间
*
* @return
*/
LocalDateTime nextTime();
/**
* 当前执行时间
*
* @return
*/
LocalDateTime theTime();
}

View File

@@ -0,0 +1,79 @@
package net.tccn.timer.scheduled;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalUnit;
/**
* @author: liangxianyou at 2018/8/5 18:05.
*/
public class ScheduledCycle implements Scheduled {
private LocalDateTime theTime;
private long period;
private TemporalUnit unit = ChronoUnit.MILLIS;
private ScheduledCycle() {
}
public static Scheduled of(String periodCfg) {
TemporalUnit unit = ChronoUnit.MILLIS;
String endchar = "";
long period;
if (periodCfg.matches("^\\d+[y,M,d,H,m,s]$")) {
endchar = periodCfg.substring(periodCfg.length() - 1);
period = Long.parseLong(periodCfg.substring(0, periodCfg.length() - 1));
} else if (periodCfg.matches("^\\d+$")) {
period = Long.parseLong(periodCfg);
if (period <= 0) {
throw new IllegalArgumentException(String.format("ScheduledCycle period config error: [%s]", periodCfg));
}
} else {
throw new IllegalArgumentException(String.format("ScheduledCycle period config error: [%s]", periodCfg));
}
if ("y".equals(endchar)) unit = ChronoUnit.YEARS;
else if ("M".equals(endchar)) unit = ChronoUnit.MONTHS;
else if ("d".equals(endchar)) unit = ChronoUnit.DAYS;
else if ("H".equals(endchar)) unit = ChronoUnit.HOURS;
else if ("m".equals(endchar)) unit = ChronoUnit.MINUTES;
else if ("s".equals(endchar)) unit = ChronoUnit.SECONDS;
return of(period, unit);
}
public static Scheduled of(long period) {
return of(period, ChronoUnit.MILLIS);
}
public static Scheduled of(long period, TemporalUnit unit) {
LocalDateTime theTime = LocalDateTime.now().plus(period, unit);
return of(theTime, period, unit);
}
public static Scheduled of(LocalDateTime startTime, long period) {
return of(startTime, period, ChronoUnit.MILLIS);
}
public static Scheduled of(LocalDateTime startTime, long period, TemporalUnit unit) {
ScheduledCycle scheduled = new ScheduledCycle();
scheduled.theTime = startTime;
scheduled.period = period;
scheduled.unit = unit;
return scheduled;
}
@Override
public LocalDateTime nextTime() {
if (theTime.isAfter(LocalDateTime.now())) {
return theTime;
}
return theTime = theTime.plus(period, unit);
}
@Override
public LocalDateTime theTime() {
return theTime;
}
}

View File

@@ -0,0 +1,499 @@
package net.tccn.timer.scheduled;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
/**
* 时间解析器
*
* @author: liangxianyou
*/
@SuppressWarnings("Duplicates")
public class ScheduledExpres implements Scheduled {
private int year;
private int month;
private int[] minutes;
private int[] hours;
private int[] days;
private int[] monthes;
private int[] weeks;
private String cfg;
private String[] cfgArr;
private LocalDateTime theTime;
private int _y, _M, _d, _H, _m;
@Deprecated
private ScheduledExpres(String cfg) {
this.cfg = cfg;
this.theTime = LocalDateTime.now();
initTheTime();
}
@Deprecated
private ScheduledExpres(final LocalDateTime startTime, String cfg) {
LocalDateTime now = LocalDateTime.now();
this.theTime = now.isAfter(startTime) ? now : startTime;
this.cfg = cfg;
initTheTime();
}
public static Scheduled of(String cfg) {
return new ScheduledExpres(cfg);
}
public static Scheduled of(final LocalDateTime startTime, String cfg) {
return new ScheduledExpres(startTime, cfg);
}
//寻找初始合法时间
public void initTheTime() {
year = theTime.getYear();
month = theTime.getMonthValue();
cfgArr = cfg.split(" ");
setWeeks();
setMonthes();
setDays();
setHours();
setMinutes();
_y = theTime.getYear();
_M = theTime.getMonthValue();
_d = theTime.getDayOfMonth();
_H = theTime.getHour();
_m = theTime.getMinute();
String cmd = "";//y M d H m
if (days.length == 0) cmd = "M";
do {
carry(cmd);
int inx;
if ((inx = nowOk(monthes, _M)) < 0) {
cmd = "y";
continue;
}
_M = monthes[inx];
if ((inx = nowOk(days, _d)) < 0) {
cmd = "M";
continue;
}
_d = days[inx];
if ((inx = nowOk(hours, _H)) < 0) {
cmd = "d";
continue;
}
_H = hours[inx];
if ((inx = nowOk(minutes, _m)) < 0) {
cmd = "H";
continue;
}
_m = minutes[inx];
break;
} while (true);
theTime = LocalDateTime.of(_y, _M, _d, _H, _m);
LocalDateTime now = LocalDateTime.now();
while (theTime.isBefore(now)) {
theTime = carry("m");
}
}
/**
* 下一次执行的时间
*
* @return
*/
@Override
public LocalDateTime nextTime() {
if (theTime.isAfter(LocalDateTime.now())) {
return theTime;
}
return theTime = carry("m");
}
@Override
public LocalDateTime theTime() {
return theTime;
}
/**
* 通过发送指令进行进位
*
* @param cmd 进位指令
*/
private LocalDateTime carry(String cmd) {
int inx;
while (!"".equals(cmd)) {
switch (cmd) {
case "y":
_y = this.year = ++_y;
_M = this.month = monthes[0];
setDays();
if (days.length == 0) {
cmd = "M";
continue;
}
_d = days[0];
_H = hours[0];
_m = minutes[0];
break;
case "M":
if (_M < monthes[0]) {
_M = monthes[0];
break;
}
inx = Arrays.binarySearch(monthes, _M);
if (inx < 0 || inx >= monthes.length - 1) {
cmd = "y";
continue;
}
_M = this.month = monthes[inx + 1];
setDays();
if (days.length == 0) {
cmd = "M";
continue;
}
_d = days[0];
_H = hours[0];
_m = minutes[0];
break;
case "d":
if (_d < days[0]) {
_d = days[0];
break;
}
inx = Arrays.binarySearch(days, _d);
if (inx < 0 || inx >= days.length - 1) {
cmd = "M";
continue;
}
_d = days[inx + 1];
_H = hours[0];
_m = minutes[0];
break;
case "H":
if (_H < hours[0]) {
_H = hours[0];
break;
}
inx = Arrays.binarySearch(hours, _H);
if (inx < 0 || inx >= hours.length - 1) {
cmd = "d";
continue;
}
_H = hours[inx + 1];
_m = minutes[0];
break;
case "m":
if (_m < minutes[0]) {
_m = minutes[0];
break;
}
inx = Arrays.binarySearch(minutes, _m);
if (inx < 0 || inx >= minutes.length - 1) {
cmd = "H";
continue;
}
_m = minutes[inx + 1];
break;
}
cmd = "";
}
return LocalDateTime.of(_y, _M, _d, _H, _m);
}
/**
* 得到初始合法时间的索引
*
* @param arr 合法时间序列
* @param n 初始选中值
* @return 合法时间的索引
*/
private int nowOk(int[] arr, int n) {
if (arr == null || arr.length == 0) return -1;
if (arr[0] > n)
return 0;
if (arr[arr.length - 1] < n)
return 0;
if (arr[arr.length - 1] == n)
return arr.length - 1;
for (int i = 0; i < arr.length - 1; i++) {
if ((arr[i] < n && arr[i + 1] > n) || arr[i] == n) {
return i;
}
}
return -1;
}
/**
* 以下为 初始化合法时间 weeksmonthesdayshourminutes 序列
*/
private void setMinutes() {
String cfg = cfgArr[0];
if ("*".equals(cfg)) {//*
minutes = new int[60];
for (int i = 0; i < 60; i++) {
minutes[i] = i;
}
} else if (cfg.matches("^[0-5]??[0-9]??$")) {//n
minutes = new int[1];
minutes[0] = Integer.parseInt(cfg);
} else if (cfg.matches("^[*]/[0-9]+$")) {// */5
String[] strArr = cfg.split("/");
int p = Integer.parseInt(strArr[1]);
minutes = new int[60 / p];
for (int i = 0; i < minutes.length; i++) {
minutes[i] = i * p;
}
} else if (cfg.matches("^([0-5]??[0-9]??,)+([0-5]??[0-9]??)?$")) {//1,3
String[] strings = cfg.split(",");
minutes = new int[strings.length];
for (int i = 0; i < strings.length; i++) {
minutes[i] = Integer.parseInt(strings[i]);
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??$")) {//1-3
String[] split = cfg.split("-");
int s = Integer.parseInt(split[0]);
int e = Integer.parseInt(split[1]);
minutes = new int[e - s + 1];
for (int i = 0; i < minutes.length; i++) {
minutes[i] = s + i;
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??/[0-5]??[0-9]??$")) {//3-18/5
String[] strArr = cfg.split("/");
String[] str2Arr = strArr[0].split("-");
int s = Integer.parseInt(str2Arr[0]);
int e = Integer.parseInt(str2Arr[1]);
int p = Integer.parseInt(strArr[1]);
minutes = new int[(e - s) / p];
for (int i = 0; i < minutes.length; i++) {
minutes[i] = s + i * p;
}
}
}
private void setHours() {
String cfg = cfgArr[1];
if ("*".equals(cfg)) {//*
hours = new int[24];
for (int i = 0; i < hours.length; i++) {
hours[i] = i;
}
} else if (cfg.matches("^[0-5]??[0-9]??$")) {//n
hours = new int[1];
hours[0] = Integer.parseInt(cfg);
} else if (cfg.matches("^[*]/[0-9]+$")) {// */5
String[] strArr = cfg.split("/");
int p = Integer.parseInt(strArr[1]);
hours = new int[24 / p];
for (int i = 0; i < hours.length; i++) {
hours[i] = i * p;
}
} else if (cfg.matches("^([0-5]??[0-9]??,)+([0-5]??[0-9]??)?$")) {//1,3
String[] strArr = cfg.split(",");
hours = new int[strArr.length];
for (int i = 0; i < strArr.length; i++) {
hours[i] = Integer.parseInt(strArr[i]);
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??$")) {//1-3
String[] split = cfg.split("-");
int s = Integer.parseInt(split[0]);
int e = Integer.parseInt(split[1]);
hours = new int[e - s + 1];
for (int i = 0; i < hours.length; i++) {
hours[i] = s + i;
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??/[0-5]??[0-9]??$")) {//3-18/5
String[] strArr = cfg.split("/");
String[] str2Arr = strArr[0].split("-");
int s = Integer.parseInt(str2Arr[0]);
int e = Integer.parseInt(str2Arr[1]);
int p = Integer.parseInt(strArr[1]);
hours = new int[(e - s) / p];
for (int i = 0; i < hours.length; i++) {
hours[i] = s + i * p;
}
}
}
private void setWeeks() {
String cfg = cfgArr[4];
if ("*".equals(cfg)) {//*
weeks = new int[7];
for (int i = 0; i < weeks.length; i++) {
weeks[i] = i + 1;
}
} else if (cfg.matches("^[0-5]??[0-9]??$")) {//n
weeks = new int[1];
weeks[0] = Integer.parseInt(cfg);
} else if (cfg.matches("^[*]/[0-9]+$")) {// */5
String[] strArr = cfg.split("/");
int p = Integer.parseInt(strArr[1]);
weeks = new int[7 / p];
for (int i = 0; i < weeks.length; i++) {
weeks[i] = i * p;
}
} else if (cfg.matches("^([0-5]??[0-9]??,)+([0-5]??[0-9]??)?$")) {//1,3
String[] strArr = cfg.split(",");
weeks = new int[strArr.length];
for (int i = 0; i < strArr.length; i++) {
weeks[i] = Integer.parseInt(strArr[i]);
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??$")) {//1-3
String[] split = cfg.split("-");
int s = Integer.parseInt(split[0]);
int e = Integer.parseInt(split[1]);
weeks = new int[e - s + 1];
for (int i = 0; i < weeks.length; i++) {
weeks[i] = s + i;
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??/[0-5]??[0-9]??$")) {//3-18/5
String[] strArr = cfg.split("/");
String[] str2Arr = strArr[0].split("-");
int s = Integer.parseInt(str2Arr[0]);
int e = Integer.parseInt(str2Arr[1]);
int p = Integer.parseInt(strArr[1]);
weeks = new int[(e - s) / p];
for (int i = 0; i < weeks.length; i++) {
weeks[i] = s + i * p;
}
}
}
private void setMonthes() {
String cfg = cfgArr[3];
if ("*".equals(cfg)) {//*
monthes = new int[12];
for (int i = 0; i < monthes.length; i++) {
monthes[i] = i + 1;
}
} else if (cfg.matches("^[0-5]??[0-9]??$")) {//n
monthes = new int[1];
monthes[0] = Integer.parseInt(cfg);
} else if (cfg.matches("^[*]/[0-9]+$")) {// */5
String[] strArr = cfg.split("/");
int p = Integer.parseInt(strArr[1]);
monthes = new int[12 / p];
for (int i = 0; i < monthes.length; i++) {
monthes[i] = i * p;
}
} else if (cfg.matches("^([0-5]??[0-9]??,)+([0-5]??[0-9]??)?$")) {//1,3
String[] strArr = cfg.split(",");
monthes = new int[strArr.length];
for (int i = 0; i < strArr.length; i++) {
monthes[i] = Integer.parseInt(strArr[i]);
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??$")) {//1-3
String[] split = cfg.split("-");
int s = Integer.parseInt(split[0]);
int e = Integer.parseInt(split[1]);
monthes = new int[e - s + 1];
for (int i = 0; i < monthes.length; i++) {
monthes[i] = s + i;
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??/[0-5]??[0-9]??$")) {//3-18/5
String[] strArr = cfg.split("/");
String[] str2Arr = strArr[0].split("-");
int s = Integer.parseInt(str2Arr[0]);
int e = Integer.parseInt(str2Arr[1]);
int p = Integer.parseInt(strArr[1]);
monthes = new int[(e - s) / p];
for (int i = 0; i < monthes.length; i++) {
monthes[i] = s + i * p;
}
}
}
private void setDays() {
String cfg = cfgArr[2];
//当前月份总天数,
LocalDate firstDay = LocalDate.of(year, month, 1);
int lengthOfMonth = firstDay.lengthOfMonth();
if ("*".equals(cfg)) {//*
days = new int[lengthOfMonth];
for (int i = 0; i < days.length; i++) {
days[i] = i + 1;
}
} else if (cfg.matches("^[0-5]??[0-9]??$")) {//n
days = new int[1];
days[0] = Integer.parseInt(cfg);
} else if (cfg.matches("^[*]/[0-9]+$")) {// */5
String[] strArr = cfg.split("/");
int p = Integer.parseInt(strArr[1]);
days = new int[lengthOfMonth / p];
for (int i = 0; i < days.length; i++) {
days[i] = i * p;
}
} else if (cfg.matches("^([0-5]??[0-9]??,)+([0-5]??[0-9]??)?$")) {//1,3
String[] strArr = cfg.split(",");
days = new int[strArr.length];
for (int i = 0; i < strArr.length; i++) {
days[i] = Integer.parseInt(strArr[i]);
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??$")) {//1-3
String[] split = cfg.split("-");
int s = Integer.parseInt(split[0]);
int e = Integer.parseInt(split[1]);
days = new int[e - s + 1];
for (int i = 0; i < days.length; i++) {
days[i] = s + i;
}
} else if (cfg.matches("^[0-5]??[0-9]??\\-[0-5]??[0-9]??/[0-5]??[0-9]??$")) {//3-18/5
String[] strArr = cfg.split("/");
String[] str2Arr = strArr[0].split("-");
int s = Integer.parseInt(str2Arr[0]);
int e = Integer.parseInt(str2Arr[1]);
int p = Integer.parseInt(strArr[1]);
days = new int[(e - s) / p];
for (int i = 0; i < days.length; i++) {
days[i] = s + i * p;
}
}
int firstWeek = firstDay.getDayOfWeek().getValue();
List<Integer> allDay = new ArrayList<>();
for (int i = 0; i < days.length; i++) {
//int week = 7 - Math.abs(i - firstWeek) % 7;//当前星期X
int week;
int d = days[i];
if (d + firstWeek <= 8) {
week = firstWeek + d - 1;
} else {
week = (d - (8 - firstWeek)) % 7;
if (week == 0) week = 7;
}
//System.out.printf("M:%s,d:%s,w:%s%n", month, d, week);
if (Arrays.binarySearch(weeks, week) > -1) {
allDay.add(d);//加入日期
}
}
days = new int[allDay.size()];
for (int i = 0; i < allDay.size(); i++) {
days[i] = allDay.get(i);
}
}
}

View File

@@ -0,0 +1,14 @@
package net.tccn.timer.task;
/**
* @author: liangxianyou at 2018/12/8 17:24.
*/
@FunctionalInterface
public interface Job {
/**
* 任务执行的内容
*/
void execute(Task task);
}

View File

@@ -0,0 +1,69 @@
package net.tccn.timer.task;
import net.tccn.timer.TimerExecutor;
import net.tccn.timer.scheduled.Scheduled;
/**
* @author: liangxianyou at 2018/8/5 19:32.
*/
public interface Task extends Runnable {
/**
* 得到任务名称
*
* @return
*/
String getName();
/**
* 设置任务执行计划
*
* @param scheduled
*/
void setScheduled(Scheduled scheduled);
/**
* 得到下一次执行计划的时间并设置thenTime
*
* @return
*/
long nextTime();
/**
* 任务即将执行的时间点
*
* @return
*/
long theTime();
/**
* 是否完成
*
* @return
*/
boolean isComplete();
/**
* 完成任务(结束标记)
*
* @param complete
*/
void setComplete(boolean complete);
/**
* 开始时间(创建时间)
*
* @return
*/
long startTime();
TimerExecutor getTimerExecutor();
void setTimerExecutor(TimerExecutor timerExecutor);
/**
* 得到总执行次数
* @return
*/
int getExecCount();
}