多线程将DB数据缓存到Elasticsearch里

╰半夏微凉° 2021-12-05 03:43 222阅读 0赞

使用多线程将库存信息同步到Elasticsearch缓存里，代码如下：

@GetMapping("/internal/unifiedInventory/sync/dbToEs")
        public void syncUnifiedInventoryDbToEs() {
            ExecutorService executors = Executors.newFixedThreadPool(20);
            Date date = DateUtils.addDays(new Date(), -1);
            List<UnifiedInventoryRecord> inventoryRecordList = unifiedInventoryRecordRepository.findByCreateTimeAfter(FormatDateUtil.time0000(date));
            if (CollectionUtil.isEmpty(inventoryRecordList)) {
                return;
            }
            List<Long> idLIst = inventoryRecordList.stream().map(UnifiedInventoryRecord::getUnifiedInventoryId).distinct().collect(Collectors.toList());
            for (List<Long> ids:partition(idLIst,900)){
                executors.execute(new Runnable() {
                    @Override
                    public void run() {
                        List<UnifiedInventory> dbList = unifiedInventoryJpaRepository.findAll(ids);
                        List<UnifiedInventoryEs> esList = CommonUntils.iterableToList(unifiedInventoryRepository.findAll(ids));
                        Map<Long,UnifiedInventoryEs> map = esList.stream().collect(Collectors.toMap(UnifiedInventoryEs::getUnifiedInventoryId, Function.identity()));
                        for (UnifiedInventory unifiedInventory : dbList) {
                            UnifiedInventoryEs unifiedInventoryEs = map.get(unifiedInventory.getUnifiedInventoryId());
                            if(Objects.isNull(unifiedInventoryEs)){
                                log.info("unifidInventory Es 缺失id:{}",unifiedInventory.getUnifiedInventoryId());
                                unifiedInventoryService.syncUnifiedInventory(new UnifiedInventorySyncLock(unifiedInventory.getUnifiedInventoryId()));
                            }else if(!unifiedInventoryEs.getCount().equals(unifiedInventory.getCount())){
                                log.info("unifidInventory数量不一致id:{}",unifiedInventory.getUnifiedInventoryId());
                                unifiedInventoryEs.setCount(unifiedInventory.getCount());
                                unifiedInventoryRepository.save(unifiedInventoryEs);
                            }
                        }
                    }
                });
    
            }
            executors.shutdown();
        }