Merge branch 'feature' into 'release'

Feature

See merge request !238

src/main/java/com/zhiwei/brandkbs2/service/impl/TaskServiceImpl.java

@@ -116,15 +116,15 @@ public class TaskServiceImpl implements TaskService {
         }
         // 替换成新的记录map
         channelIndexRecordMap = newRecordMap;
-//        ListUtils.partition(insertList, 1000).forEach(list -> {
-//            channelDao.insertMany(list);
-//        });
+        ListUtils.partition(insertList, 1000).forEach(list -> {
+            channelDao.insertMany(list);
+        });
         log.info("渠道统计-渠道总计-录入完毕,新增渠道{}条,更新渠道{}条", insertList.size(), channelIndexRecordMap.size() - insertList.size());
         // 获得单位时间内最小最大时间戳
         Long[] timeMinMax = Tools.timeMinMax(rangeTimeRecords.stream().map(Pair::getLeft).collect(Collectors.toList()));
         List<ChannelRecord> channelRecords = ChannelRecord.createChannelRecords(timeMinMax[0], timeMinMax[1], channelIndexRecordMap);
-//        channelEsDao.upsertChannelRecord(channelRecords);
-        // 同步channelCopy
+        channelEsDao.upsertChannelRecord(channelRecords);
+        // 同步ES-channelCopy,区分insertList和updateList
         ListUtils.partition(insertList, 1000).forEach(list -> {
             channelEsDao.batchInsert(list.stream().map(Channel::createChannelCopyMap).collect(Collectors.toList()));
         });
@@ -207,7 +207,7 @@ public class TaskServiceImpl implements TaskService {
                 channel.setRecord(entry.getValue());
                 Channel updateChannel = channelService.calculateChannelEmotionIndex(channel);
                 updateChannel.updateInfo(entry.getKey(), entry.getValue());
-//                channelDao.updateOne(updateChannel);
+                channelDao.updateOne(updateChannel);
                 updateList.add(channel);
             }
             // 设置查询数值