数据批处理队列方式
目录
数据批处理（队列方式）
数据批处理（队列方式）
public class DataProcessor {
private static final int THREAD_COUNT = 4;
private static final int QUEUE_SIZE = 10;
private LinkedBlockingQueue<Data> queue = new LinkedBlockingQueue<>(QUEUE_SIZE);
public DataProcessor() {
ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
for (int i = 0; i < THREAD_COUNT; i++) {
executor.execute(() -> {
try {
while (true) {
Data data = queue.take();
processData(data);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
}
public void addData(Data data) {
try {
queue.put(data);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void processData(Data data) {
// Process data here
System.out.println("Processing data: " + data);
}
public static void main(String[] args) {
DataProcessor processor = new DataProcessor();
// Add data to be processed
for (int i = 0; i < 20; i++) {
Data data = new Data("Data " + i);
processor.addData(data);
}
}
static class Data {
private String value;
public Data(String value) {
this.value = value;
}
@Override
public String toString() {
return value;
}
}
}参考网址：
一：https://blog.csdn.net/zhizhengguan/article/details/86551270
二：https://blog.csdn.net/qq_41128049/article/details/134442487
数据批处理入库（线程池方式）
public void reSync() {
// 超过半小时状态为1的数据状态重置0
Date date = DateUtil.getAddMinuteDate(new Date(), -30);
baseMapper.resetTimeoutStatus(date);
// 查询推送失败且失败次数小于6次的
List<CardRecordSync> list = baseMapper.listFail();
if (list.isEmpty()) {
return;
}
log.info("待处理补推记录数size={}", list.size());
List<List<CardRecordSync>> partition = Lists.partition(list, 100);
for (List<CardRecordSync> syncs : partition) {
executor.execute(() -> {
List<Long> ids = syncs.stream().map(CardRecordSync::getId).collect(Collectors.toList());
// 状态变更为处理中
EntityWrapper<CardRecordSync> wrapper = new EntityWrapper<>();
wrapper.in("ID", ids);
CardRecordSync po = new CardRecordSync();
po.setStatus(CardRecordSyncStatus.PUSHING.getStatus());
baseMapper.update(po, wrapper);
syncs.forEach(record -> {
// boolean flag = thirdService.cardRecordSync(record.getSyncParam());
DataSyncResult dataSyncResult = thirdService.cardRecordSync(record.getSyncParam());
boolean flag = dataSyncResult.isDataSyncSuc();
int status = CardRecordSyncStatus.FAIL.getStatus();
if (flag) {
status = CardRecordSyncStatus.SUCCESS.getStatus();
}
baseMapper.updateStatusDescById(dataSyncResult.getDataSyncSucDesc(),status, record.getId());
});
});
}
}