思路
1.首先设计一张表,记录任务状态,执行时间,已执行,未执行等等
其次任务来的时候写入表中,任务标识未执行
如果此时有1000个任务都过来了,那么表中也有1000行记录
接着JVM内存溢出OOM,程序挂了,未执行的任务就知道是哪些了.
系统再次启动的时候,依次读取未执行的任务加入线程池中复盘,二次执行
2.对正在处理和阻塞队列的任务或者阻塞队列进行任务的持久化操作，
当断电或者系统宕机无法继续操作下去的话，可以通过回溯日志来处理已经执行成功的操作，重新执行整个阻塞队列。
3.实际业务场景
多线程处理数据推送，更新数据状态由0->1，同时记录执行到的位置id的日志
根据当前处理到数据库的位置，重启机器后传入数据id的位置重新加载推送
4.代码推送
Long startId = maxMinIdEntity.getMinId().longValue();
Long endId = startId.longValue() + QConfigUtil.getIntValue(Constants.PushDataToTpLimitCount, 1000);
long executeCount = 0;
while (startId.compareTo(maxMinIdEntity.getMaxId()) <= 0) {
List<T> list = getDataFromDb(date, startId, endId);
Clog.infoByTags("推送数据", String.format("获取数据：%s-%s;Count:%s", startId, endId, list.size()), mapLog);
if (CollectionUtils.isNotEmpty(list)) {
for (T t : list) {
TpEntity<R> tpEntity = new TpEntity<>();
tpEntity.setInterfaceName(getInterfaceName());
tpEntity.setData(Arrays.asList(convertToTpData(t)));
QmqProducer.sendMessageToTp(getQmqSubject(), tpEntity);
//更新数据状态 可以知道当前推送到了哪一行数据
int excuteCount = updatePushResult(t);
Clog.infoByTags("推送数据", String.format("推送Qmq,更新DB结果:%s", excuteCount), mapLog);
}
executeCount += list.size();
}
链接：https://blog.csdn.net/weixin_37961431/article/details/108230314