lanyuanxiaoyao/leopard
1
0
Fork 0
Code Activity

feat: 并行处理任务

Browse Source
This commit is contained in:
兰缘小妖
2025-09-07 21:26:58 +08:00
parent 9e2d2e3845
commit 769bd4a90b
2 changed files with 60 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
leopard-server/src/main/java/com/lanyuanxiaoyao/leopard/server/service/task/CheckDailyNode.java
View File

@@ -41,7 +41,8 @@ public class CheckDailyNode extends NodeComponent {
.map(item -> LocalDate.parse(item.get(0), TuShareService.TRADE_FORMAT)) .map(item -> LocalDate.parse(item.get(0), TuShareService.TRADE_FORMAT))
.filter(date -> date.isBefore(nowDate) || date.isEqual(nowDate)) .filter(date -> date.isBefore(nowDate) || date.isEqual(nowDate))
.toList(); .toList();
for (Stock stock : stocks) { stocks.parallelStream()
.forEach(stock -> {
if (exchange.equals(stock.getMarket())) { if (exchange.equals(stock.getMarket())) {
var existsTradeDates = dailyService.findDistinctTradeDateByStockId(stock.getId()); var existsTradeDates = dailyService.findDistinctTradeDateByStockId(stock.getId());
var missedTradeDates = allTradeDates.stream() var missedTradeDates = allTradeDates.stream()
@@ -56,7 +57,7 @@ public class CheckDailyNode extends NodeComponent {
)); ));
} }
} }
} });
} }
if (ObjectUtil.isNotEmpty(reports)) { if (ObjectUtil.isNotEmpty(reports)) {
var context = getContextBean(TaskMonitorNodes.TaskMonitorContext.class); var context = getContextBean(TaskMonitorNodes.TaskMonitorContext.class);

30
leopard-server/src/main/java/com/lanyuanxiaoyao/leopard/server/service/task/UpdateDailyNode.java
View File

@@ -46,14 +46,14 @@ public class UpdateDailyNode extends NodeComponent {
} }
var existsTradeDates = dailyService.findDistinctTradeDate(); var existsTradeDates = dailyService.findDistinctTradeDate();
var nowDate = LocalDate.now(); var nowDate = LocalDate.now();
var targetDates = tradeDates.stream() var stocks = stockService.list();
var stocksMap = stocks.stream().collect(Collectors.toMap(Stock::getCode, stock -> stock));
tradeDates.stream()
.filter(date -> date.isBefore(nowDate) || date.isEqual(nowDate)) .filter(date -> date.isBefore(nowDate) || date.isEqual(nowDate))
.filter(date -> !existsTradeDates.contains(date)) .filter(date -> !existsTradeDates.contains(date))
.sorted() .sorted()
.toList(); .parallel()
var stocks = stockService.list(); .forEach(tradeDate -> {
var stocksMap = stocks.stream().collect(Collectors.toMap(Stock::getCode, stock -> stock));
for (LocalDate tradeDate : targetDates) {
var factorResponse = tuShareService.factorList(tradeDate); var factorResponse = tuShareService.factorList(tradeDate);
var factorMap = new HashMap<String, Double>(); var factorMap = new HashMap<String, Double>();
for (List<String> item : factorResponse.data().items()) { for (List<String> item : factorResponse.data().items()) {
@@ -72,15 +72,15 @@ public class UpdateDailyNode extends NodeComponent {
var factor = factorMap.get(code); var factor = factorMap.get(code);
var daily = new Daily(); var daily = new Daily();
daily.setTradeDate(LocalDate.parse(item.get(1), TuShareService.TRADE_FORMAT)); daily.setTradeDate(LocalDate.parse(item.get(1), TuShareService.TRADE_FORMAT));
daily.setOpen(item.get(2) == null ? null : Double.valueOf(item.get(2))); daily.setOpen(item.get(2) == null ? null : Double.parseDouble(item.get(2)));
daily.setHigh(item.get(3) == null ? null : Double.valueOf(item.get(3))); daily.setHigh(item.get(3) == null ? null : Double.parseDouble(item.get(3)));
daily.setLow(item.get(4) == null ? null : Double.valueOf(item.get(4))); daily.setLow(item.get(4) == null ? null : Double.parseDouble(item.get(4)));
daily.setClose(item.get(5) == null ? null : Double.valueOf(item.get(5))); daily.setClose(item.get(5) == null ? null : Double.parseDouble(item.get(5)));
daily.setPreviousClose(item.get(6) == null ? null : Double.valueOf(item.get(6))); daily.setPreviousClose(item.get(6) == null ? null : Double.parseDouble(item.get(6)));
daily.setPriceChangeAmount(item.get(7) == null ? null : Double.valueOf(item.get(7))); daily.setPriceChangeAmount(item.get(7) == null ? null : Double.parseDouble(item.get(7)));
daily.setPriceFluctuationRange(item.get(8) == null ? null : Double.valueOf(item.get(8))); daily.setPriceFluctuationRange(item.get(8) == null ? null : Double.parseDouble(item.get(8)));
daily.setVolume(item.get(9) == null ? null : Double.valueOf(item.get(9))); daily.setVolume(item.get(9) == null ? null : Double.parseDouble(item.get(9)));
daily.setTurnover(item.get(10) == null ? null : Double.valueOf(item.get(10))); daily.setTurnover(item.get(10) == null ? null : Double.parseDouble(item.get(10)) * 1000);
daily.setFactor(factor); daily.setFactor(factor);
daily.setStock(stock); daily.setStock(stock);
dailyService.save(daily); dailyService.save(daily);
@@ -94,6 +94,6 @@ public class UpdateDailyNode extends NodeComponent {
return false; return false;
} }
}); });
} });
} }
} }