CanalScheduling.java 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478
  1. package com.ruoyi.web.job;
  2. import cn.hutool.core.bean.BeanUtil;
  3. import cn.hutool.core.bean.copier.CopyOptions;
  4. import cn.hutool.core.collection.CollUtil;
  5. import cn.hutool.core.convert.Convert;
  6. import cn.hutool.core.date.DateUtil;
  7. import cn.hutool.core.util.NumberUtil;
  8. import cn.hutool.core.util.ObjectUtil;
  9. import cn.hutool.core.util.StrUtil;
  10. import cn.hutool.http.HttpUtil;
  11. import com.alibaba.fastjson.JSON;
  12. import com.alibaba.fastjson.JSONArray;
  13. import com.alibaba.fastjson.JSONObject;
  14. import com.alibaba.otter.canal.client.CanalConnector;
  15. import com.alibaba.otter.canal.client.CanalConnectors;
  16. import com.alibaba.otter.canal.protocol.CanalEntry;
  17. import com.alibaba.otter.canal.protocol.Message;
  18. import com.google.protobuf.InvalidProtocolBufferException;
  19. import com.ruoyi.common.constant.CacheConstants;
  20. import com.ruoyi.common.constant.ElasticConstants;
  21. import com.ruoyi.common.core.redis.RedisCache;
  22. import com.ruoyi.common.utils.JdbcTypeUtil;
  23. import com.ruoyi.common.utils.uuid.IdUtils;
  24. import com.ruoyi.framework.config.ElasticSearchClient;
  25. import com.ruoyi.system.domain.IllegalShipData;
  26. import com.ruoyi.system.domain.vo.AisShipInfo;
  27. import com.ruoyi.system.domain.vo.ShipEepReportRecInfo;
  28. import com.ruoyi.system.service.IAisInfoService;
  29. import com.ruoyi.system.service.IIllegalShipDataService;
  30. import com.ruoyi.system.service.ISysConfigService;
  31. import com.ruoyi.web.core.config.CanalConfig;
  32. import lombok.SneakyThrows;
  33. import lombok.extern.slf4j.Slf4j;
  34. import org.springframework.beans.BeansException;
  35. import org.springframework.beans.factory.annotation.Autowired;
  36. import org.springframework.beans.factory.annotation.Value;
  37. import org.springframework.context.ApplicationContext;
  38. import org.springframework.context.ApplicationContextAware;
  39. import org.springframework.scheduling.annotation.Scheduled;
  40. import org.springframework.stereotype.Component;
  41. import org.springframework.util.LinkedCaseInsensitiveMap;
  42. import java.net.InetSocketAddress;
  43. import java.util.ArrayList;
  44. import java.util.HashMap;
  45. import java.util.List;
  46. import java.util.Map;
  47. /**
  48. * @Description: TODO
  49. * @Author: huangcheng
  50. * @Date: 2021/8/16
  51. * @Version V1.0
  52. */
  53. @Slf4j
  54. @Component
  55. public class CanalScheduling implements Runnable, ApplicationContextAware {
  56. private ApplicationContext applicationContext;
  57. @Autowired
  58. private ElasticSearchClient client;
  59. @Autowired
  60. private RedisCache redisCache;
  61. @Autowired
  62. private ISysConfigService configService;
  63. @Autowired
  64. private IAisInfoService aisInfoService;
  65. // @Resource
  66. // private CanalConnector canalConnector;
  67. @Value("${black.snapImgUrl}")
  68. private String blackImgUrl;
  69. @Autowired
  70. private IIllegalShipDataService illegalShipDataService;
  71. @Autowired
  72. private CanalConfig canalConfig;
  73. @Value("${spring.profiles.active}")
  74. private String env;
  75. public final static Map<String, String> orgMap = new HashMap<String, String>() {{
  76. put("南京三桥", "南京海事局");
  77. put("南京四桥", "南京海事局");
  78. put("润扬大桥", "扬州海事局");
  79. put("泰州大桥", "泰州海事局");
  80. put("江阴大桥", "江阴海事局");
  81. put("苏通大桥", "常熟海事局");
  82. }};
  83. public final static Map<String, String> heiyanDeviceMap = new HashMap<String, String>() {{
  84. put("南京三桥", "65");
  85. put("润扬大桥", "66");
  86. put("江阴大桥", "67");
  87. put("苏通大桥", "68");
  88. }};
  89. @Override
  90. @Scheduled(fixedDelay = 100) //每隔100秒执行
  91. public void run() {
  92. for (CanalConfig.Config config : canalConfig.getConfigs()) {
  93. // 创建链接
  94. CanalConnector canalConnector = CanalConnectors
  95. .newSingleConnector(new InetSocketAddress(config.getHostname(), config.getPort()), config.getDestination(), "", "");
  96. long batchId = -1;
  97. try {
  98. canalConnector.connect();
  99. canalConnector.subscribe();
  100. canalConnector.rollback();
  101. //每次拉取条数
  102. int batchSize = 1000;
  103. Message message = canalConnector.getWithoutAck(batchSize);
  104. //批次id
  105. batchId = message.getId();
  106. List<CanalEntry.Entry> entries = message.getEntries();
  107. if (batchId != -1 && entries.size() > 0) {
  108. entries.forEach(entry -> {
  109. //MySQL种my.cnf中配置的是binlog_format = ROW,这里只解析ROW类型
  110. if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
  111. //解析处理
  112. publishCanalEvent(entry);
  113. }
  114. });
  115. }
  116. canalConnector.ack(batchId);
  117. } catch (Exception e) {
  118. log.info("canal存在异常:{}", e.getMessage());
  119. e.printStackTrace();
  120. canalConnector.rollback(batchId);
  121. } finally {
  122. // 断开连接
  123. canalConnector.disconnect();
  124. }
  125. }
  126. }
  127. private void publishCanalEvent(CanalEntry.Entry entry) {
  128. //表名
  129. String tableName = entry.getHeader().getTableName();
  130. //数据库名
  131. String database = entry.getHeader().getSchemaName();
  132. // 操作类型
  133. CanalEntry.EventType eventType = entry.getHeader().getEventType();
  134. log.info(String.format("========> binlog[%s:%s] , name[%s,%s] , eventType[%s]",
  135. entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
  136. database, tableName,
  137. eventType));
  138. CanalEntry.RowChange rowChange;
  139. try {
  140. rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
  141. } catch (InvalidProtocolBufferException e) {
  142. e.printStackTrace();
  143. return;
  144. }
  145. rowChange.getRowDatasList().forEach(rowData -> {
  146. //获取改变前的数据
  147. List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
  148. //获取改变后的数据
  149. List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
  150. Map<String, Object> beforeColumnsToMap = toCaseInsensitiveMap(parseColumnsToMap(beforeColumnsList, tableName));
  151. Map<String, Object> afterColumnsToMap = toCaseInsensitiveMap(parseColumnsToMap(afterColumnsList, tableName));
  152. //插入es
  153. indexES(beforeColumnsToMap, afterColumnsToMap, eventType, database, tableName);
  154. });
  155. }
  156. Map<String, Object> parseColumnsToMap(List<CanalEntry.Column> columns, String tableName) {
  157. Map<String, Object> map = new HashMap<>();
  158. columns.forEach(column -> {
  159. if (column == null) {
  160. return;
  161. }
  162. map.put(StrUtil.lowerFirst(StrUtil.toCamelCase(column.getName())), JdbcTypeUtil.typeConvert(tableName, column.getName(), column.getValue(), column.getSqlType(), column.getMysqlType()));
  163. });
  164. return map;
  165. }
  166. @SneakyThrows
  167. private void indexES(Map<String, Object> beforeDataMap, Map<String, Object> afterDataMap, CanalEntry.EventType eventType, String database, String table) {
  168. log.info("eventType:{},database:{},table:{}\nbeforeMap:{},\n afterMap:{}", eventType, database, table, beforeDataMap, afterDataMap);
  169. if (!StrUtil.equalsAnyIgnoreCase(database, "smoke_api", "ship", "ais_database")) {
  170. return;
  171. }
  172. //不是user表中的数据不处理
  173. if (!StrUtil.equalsAnyIgnoreCase(table,
  174. "ship_recognition",
  175. "ship_snap_address",
  176. "alert",
  177. "monitor_point",
  178. "device",
  179. "illegal_ship",
  180. "sem_instrument")) {
  181. return;
  182. }
  183. String so2ThresholdVal = configService.selectConfigByKey("so2.so2");
  184. String blackThresholdVal = configService.selectConfigByKey("black.rcgSoot");
  185. // 根据不同类型处理相应的逻辑
  186. switch (eventType) {
  187. case INSERT:
  188. // 黑烟船舶数据入库
  189. if (StrUtil.equalsIgnoreCase(database, "smoke_api") && StrUtil.equalsIgnoreCase(table, "ship_recognition")) {
  190. if (ObjectUtil.isEmpty(afterDataMap.get("aisMmsi")) /*|| Convert.toStr(afterDataMap.get("aisMmsi")).length() != 9*/) {
  191. break;
  192. }
  193. // 判断数据状态
  194. if (ObjectUtil.isNotEmpty(afterDataMap.get("rcgSoot")) && NumberUtil.isGreaterOrEqual(Convert.toBigDecimal(afterDataMap.get("rcgSoot")),
  195. Convert.toBigDecimal(blackThresholdVal))) {
  196. // 查询黑烟图片
  197. if ("prod".equals(env)) {
  198. String imgUrl = HttpUtil.get(blackImgUrl + Convert.toStr(afterDataMap.get("id")));
  199. JSONObject response = JSON.parseObject(imgUrl);
  200. if (200 == response.getInteger("code")) {
  201. List<String> newImgList = new ArrayList<>();
  202. JSONArray imgUrlList = response.getJSONObject("data").getJSONArray("imgUrlList");
  203. String sootImgUrl = response.getJSONObject("data").getString("sootImgUrl");
  204. for (Object o : imgUrlList) {
  205. // 只保存细节、全貌开头的图片
  206. String s = Convert.toStr(o);
  207. if (StrUtil.containsAny(s, "全貌", "细节")) {
  208. newImgList.add("http://" + s);
  209. }
  210. }
  211. afterDataMap.put("sootImgUrl", StrUtil.isNotBlank(sootImgUrl) ? "http://" + sootImgUrl : "");
  212. afterDataMap.put("allImgUrl", newImgList);
  213. }
  214. }
  215. // 超过阈值后直接判定为违规船舶
  216. afterDataMap.put("illegalStatus", 3);
  217. afterDataMap.put("uploadFlag", 0);
  218. // TODO 上报行政检查系统
  219. } else {
  220. afterDataMap.put("uploadFlag", 0);
  221. afterDataMap.put("illegalStatus", 1);
  222. }
  223. if (ObjectUtil.isNotEmpty(afterDataMap.get("aisMmsi")) && Convert.toStr(afterDataMap.get("aisMmsi")).length() == 9) {
  224. aisInfoService.getDynamicShipInfo(Convert.toStr(afterDataMap.get("aisMmsi")));
  225. }
  226. afterDataMap.put("mmsi", afterDataMap.get("aisMmsi"));
  227. afterDataMap.put("illegalType", "heiyan");
  228. // 查询检测点和船舶信息
  229. if (ObjectUtil.isNotEmpty(afterDataMap.get("aisMmsi")) && !"0".equals(Convert.toStr(afterDataMap.get("aisMmsi")))) {
  230. AisShipInfo shipInfo = aisInfoService.getShipInfo("", Convert.toStr(afterDataMap.get("aisMmsi")), "", "");
  231. if (shipInfo != null) {
  232. afterDataMap.put("shipName", shipInfo.getShipName());
  233. afterDataMap.put("shipRegionType", shipInfo.getShipRegionType());
  234. }
  235. }
  236. // 对于没有从ais系统拿到船名的情况下,直接取rcgShipName
  237. if (ObjectUtil.isEmpty(afterDataMap.get("shipName"))) {
  238. afterDataMap.put("shipName", afterDataMap.get("rcgShipName"));
  239. }
  240. if (ObjectUtil.isNotEmpty(afterDataMap.get("snapPos"))) {
  241. afterDataMap.put("orgName", orgMap.get(Convert.toStr(afterDataMap.get("snapPos"))));
  242. afterDataMap.put("deviceId", heiyanDeviceMap.get(Convert.toStr(afterDataMap.get("snapPos"))));
  243. }
  244. afterDataMap.put("createTime", afterDataMap.get("snapTimeFmt"));
  245. afterDataMap.put("monitorPointName", afterDataMap.get("snapPos"));
  246. // 保存数据
  247. saveIllegalData(afterDataMap, ElasticConstants.HEIYAN_SHIP_RECOGNITION);
  248. client.createDocument(ElasticConstants.HEIYAN_SHIP_RECOGNITION, Convert.toStr(afterDataMap.get("id")), afterDataMap);
  249. } else if (StrUtil.equalsIgnoreCase(database, "smoke_api") && StrUtil.equalsIgnoreCase(table, "ship_snap_address")) {
  250. // 删除设备和类型的统计缓存
  251. redisCache.deleteObject(CollUtil.set(false,
  252. CacheConstants.DEVICE_STATIC,
  253. CacheConstants.DEVICE_TYPE_STATIC,
  254. CacheConstants.DEVICE_LIST_STATIC));
  255. } else if (StrUtil.equalsIgnoreCase(database, "ship") && StrUtil.equalsIgnoreCase(table, "monitor_point")) {
  256. // 检测点信息
  257. client.createDocument(ElasticConstants.SO2_MONITOR_POINT, Convert.toStr(afterDataMap.get("id")), afterDataMap);
  258. redisCache.deleteObject(CacheConstants.DEVICE_STATIC);
  259. } else if (StrUtil.equalsIgnoreCase(database, "ship") && StrUtil.equalsIgnoreCase(table, "device")) {
  260. // 设备信息
  261. redisCache.deleteObject(CollUtil.set(false,
  262. CacheConstants.DEVICE_STATIC,
  263. CacheConstants.DEVICE_TYPE_STATIC,
  264. CacheConstants.DEVICE_LIST_STATIC));
  265. } else if (StrUtil.equalsIgnoreCase(database, "ship") && StrUtil.equalsIgnoreCase(table, "alert")) {
  266. // 船舶记录
  267. if ("prod".equals(env)) {
  268. if (ObjectUtil.isNotEmpty(afterDataMap.get("mmsi"))) {
  269. AisShipInfo shipInfo = aisInfoService.getShipInfo("", Convert.toStr(afterDataMap.get("mmsi")), "", "");
  270. if (shipInfo != null) {
  271. afterDataMap.put("shipName", shipInfo.getShipName());
  272. afterDataMap.put("shipRegionType", shipInfo.getShipRegionType());
  273. }
  274. }
  275. }
  276. // 判断是否大于阈值
  277. if (ObjectUtil.isNotEmpty(afterDataMap.get("so2Percent")) && NumberUtil.isGreaterOrEqual(Convert.toBigDecimal(afterDataMap.get("so2Percent")), Convert.toBigDecimal(so2ThresholdVal))) {
  278. // 查询检测点和船舶信息、船舶进出港记录
  279. if (ObjectUtil.isNotEmpty(afterDataMap.get("mmsi"))) {
  280. ShipEepReportRecInfo eepReportRecInfo = aisInfoService.getShipEepReportRecInfo(DateUtil.formatDate(DateUtil.lastWeek()), DateUtil.formatDate(DateUtil.tomorrow()), Convert.toStr(afterDataMap.get("mmsi")), "", "");
  281. if (eepReportRecInfo != null) {
  282. afterDataMap.put("destination", eepReportRecInfo.getNextPortName());
  283. afterDataMap.put("berthName", eepReportRecInfo.getBerthName());
  284. }
  285. }
  286. Map<String, Object> monitor = client.getDocById(ElasticConstants.SO2_MONITOR_POINT, Convert.toStr(afterDataMap.get("monitorPointId")), "name");
  287. if (monitor != null && ObjectUtil.equal(monitor.get("code"), 200)) {
  288. Map<String, String> data = Convert.toMap(String.class, String.class, monitor.get("data"));
  289. afterDataMap.put("monitorPointName", data.get("name"));
  290. afterDataMap.put("orgName", orgMap.get(data.get("name")));
  291. }
  292. // 嫌疑船舶
  293. afterDataMap.put("illegalStatus", 2);
  294. afterDataMap.put("uploadFlag", 0);
  295. // TODO 上报行政检查系统
  296. } else {
  297. afterDataMap.put("uploadFlag", 0);
  298. afterDataMap.put("illegalStatus", 1);
  299. }
  300. if (ObjectUtil.isNotEmpty(afterDataMap.get("mmsi")) && Convert.toStr(afterDataMap.get("mmsi")).length() == 9) {
  301. aisInfoService.getDynamicShipInfo(Convert.toStr(afterDataMap.get("mmsi")));
  302. }
  303. afterDataMap.put("illegalType", "guangpu");
  304. // 保存数据
  305. saveIllegalData(afterDataMap, ElasticConstants.SO2_ALERT);
  306. client.createDocument(ElasticConstants.SO2_ALERT, Convert.toStr(afterDataMap.get("id")), afterDataMap);
  307. } else if (StrUtil.equalsIgnoreCase(database, "ais_database") && (StrUtil.equalsIgnoreCase(table, "sem_instrument"))) {
  308. // 嗅探系统-站点信息
  309. client.createDocument(ElasticConstants.AIS_SEM_INSTRUMENT, Convert.toStr(afterDataMap.get("id")), afterDataMap);
  310. redisCache.deleteObject(CollUtil.set(false,
  311. CacheConstants.DEVICE_STATIC,
  312. CacheConstants.DEVICE_TYPE_STATIC,
  313. CacheConstants.DEVICE_LIST_STATIC));
  314. } else if (StrUtil.equalsIgnoreCase(database, "ais_database") && StrUtil.equalsIgnoreCase(table, "illegal_ship")) {
  315. // 嗅探系统-违规船舶
  316. if ("prod".equals(env)) {
  317. // 查询检测点和船舶信息、船舶进出港记录
  318. if (ObjectUtil.isNotEmpty(afterDataMap.get("shipMmsi"))) {
  319. AisShipInfo shipInfo = aisInfoService.getShipInfo("", Convert.toStr(afterDataMap.get("shipMmsi")), "", "");
  320. if (shipInfo != null) {
  321. afterDataMap.put("shipName", shipInfo.getShipName());
  322. afterDataMap.put("shipRegionType", shipInfo.getShipRegionType());
  323. }
  324. }
  325. }
  326. afterDataMap.put("mmsi", afterDataMap.get("shipMmsi"));
  327. afterDataMap.put("createTime", afterDataMap.get("peakTime"));
  328. afterDataMap.put("so2Percent", afterDataMap.get("sPercent"));
  329. if (ObjectUtil.isNotEmpty(afterDataMap.get("sPercent")) && NumberUtil.isGreaterOrEqual(Convert.toBigDecimal(afterDataMap.get("sPercent")), Convert.toBigDecimal(so2ThresholdVal))) {
  330. // 查询检测点和船舶信息、船舶进出港记录
  331. if (ObjectUtil.isNotEmpty(afterDataMap.get("mmsi"))) {
  332. ShipEepReportRecInfo eepReportRecInfo = aisInfoService.getShipEepReportRecInfo(DateUtil.formatDate(DateUtil.lastWeek()), DateUtil.formatDate(DateUtil.tomorrow()), Convert.toStr(afterDataMap.get("mmsi")), "", "");
  333. if (eepReportRecInfo != null) {
  334. afterDataMap.put("destination", eepReportRecInfo.getNextPortName());
  335. afterDataMap.put("berthName", eepReportRecInfo.getBerthName());
  336. }
  337. }
  338. afterDataMap.put("deviceId", afterDataMap.get("semId"));
  339. // 查询站点信息
  340. Map<String, Object> sem = client.getDocById(ElasticConstants.AIS_SEM_INSTRUMENT, Convert.toStr(afterDataMap.get("semId")), "name,category");
  341. if (sem != null && ObjectUtil.equal(sem.get("code"), 200)) {
  342. Map<String, String> data = Convert.toMap(String.class, String.class, sem.get("data"));
  343. afterDataMap.put("semName", data.get("name"));
  344. afterDataMap.put("monitorPointName", data.get("category"));
  345. afterDataMap.put("orgName", orgMap.get(data.get("category")));
  346. }
  347. // 嫌疑船舶
  348. afterDataMap.put("illegalStatus", 2);
  349. afterDataMap.put("uploadFlag", 0);
  350. // TODO 上报行政检查系统
  351. } else {
  352. afterDataMap.put("uploadFlag", 0);
  353. afterDataMap.put("illegalStatus", 1);
  354. }
  355. if (ObjectUtil.isNotEmpty(afterDataMap.get("mmsi")) && Convert.toStr(afterDataMap.get("mmsi")).length() == 9) {
  356. aisInfoService.getDynamicShipInfo(Convert.toStr(afterDataMap.get("mmsi")));
  357. }
  358. afterDataMap.put("illegalType", "xiutan");
  359. // 保存数据
  360. saveIllegalData(afterDataMap, ElasticConstants.AIS_ILLEGAL_SHIP);
  361. client.createDocument(ElasticConstants.AIS_ILLEGAL_SHIP, Convert.toStr(afterDataMap.get("id")), afterDataMap);
  362. }
  363. break;
  364. case UPDATE:
  365. if (StrUtil.equalsIgnoreCase(database, "smoke_api") && StrUtil.equalsIgnoreCase(table, "ship_snap_address")) {
  366. redisCache.deleteObject(CollUtil.set(false,
  367. CacheConstants.DEVICE_STATIC,
  368. CacheConstants.DEVICE_TYPE_STATIC,
  369. CacheConstants.DEVICE_LIST_STATIC));
  370. } else if (StrUtil.equalsIgnoreCase(database, "ship") && StrUtil.equalsIgnoreCase(table, "monitor_point")) {
  371. // 检测点信息
  372. client.updateDocument(ElasticConstants.SO2_MONITOR_POINT, Convert.toStr(afterDataMap.get("id")), afterDataMap);
  373. redisCache.deleteObject(CacheConstants.DEVICE_STATIC);
  374. } else if (StrUtil.equalsIgnoreCase(database, "ship") && StrUtil.equalsIgnoreCase(table, "device")) {
  375. // 设备信息
  376. redisCache.deleteObject(CollUtil.set(false,
  377. CacheConstants.DEVICE_STATIC,
  378. CacheConstants.DEVICE_TYPE_STATIC,
  379. CacheConstants.DEVICE_LIST_STATIC));
  380. } else if (StrUtil.equalsIgnoreCase(database, "ais_database") && (StrUtil.equalsIgnoreCase(table, "sem_instrument"))) {
  381. // 嗅探系统-站点信息
  382. client.updateDocument(ElasticConstants.AIS_SEM_INSTRUMENT, Convert.toStr(afterDataMap.get("id")), afterDataMap);
  383. redisCache.deleteObject(CollUtil.set(false,
  384. CacheConstants.DEVICE_STATIC,
  385. CacheConstants.DEVICE_TYPE_STATIC,
  386. CacheConstants.DEVICE_LIST_STATIC));
  387. }
  388. break;
  389. case DELETE:
  390. if (StrUtil.equalsIgnoreCase(database, "smoke_api") && StrUtil.equalsIgnoreCase(table, "ship_snap_address")) {
  391. redisCache.deleteObject(CollUtil.set(false,
  392. CacheConstants.DEVICE_STATIC,
  393. CacheConstants.DEVICE_TYPE_STATIC,
  394. CacheConstants.DEVICE_LIST_STATIC));
  395. } else if (StrUtil.equalsIgnoreCase(database, "ship") && StrUtil.equalsIgnoreCase(table, "monitor_point")) {
  396. // 检测点信息
  397. client.deleteDocument(ElasticConstants.SO2_MONITOR_POINT, Convert.toStr(afterDataMap.get("id")));
  398. redisCache.deleteObject(CacheConstants.DEVICE_STATIC);
  399. } else if (StrUtil.equalsIgnoreCase(database, "ship") && StrUtil.equalsIgnoreCase(table, "device")) {
  400. // 设备信息
  401. redisCache.deleteObject(CollUtil.set(false,
  402. CacheConstants.DEVICE_STATIC,
  403. CacheConstants.DEVICE_TYPE_STATIC,
  404. CacheConstants.DEVICE_LIST_STATIC));
  405. } else if (StrUtil.equalsIgnoreCase(database, "ais_database") && (StrUtil.equalsIgnoreCase(table, "sem_instrument"))) {
  406. // 嗅探系统-站点信息
  407. client.deleteDocument(ElasticConstants.AIS_SEM_INSTRUMENT, Convert.toStr(afterDataMap.get("id")));
  408. redisCache.deleteObject(CollUtil.set(false,
  409. CacheConstants.DEVICE_STATIC,
  410. CacheConstants.DEVICE_TYPE_STATIC,
  411. CacheConstants.DEVICE_LIST_STATIC));
  412. }
  413. break;
  414. default:
  415. break;
  416. }
  417. }
  418. private void saveIllegalData(Map<String, Object> afterDataMap, String indexName) {
  419. if (Convert.toInt(afterDataMap.get("illegalStatus")) != 1) {
  420. IllegalShipData illegalShipData = new IllegalShipData();
  421. BeanUtil.fillBeanWithMap(afterDataMap, illegalShipData, CopyOptions.create().setIgnoreProperties("id"));
  422. illegalShipData.setId(IdUtils.fastSimpleUUID());
  423. illegalShipData.setSystemId(Convert.toStr(afterDataMap.get("id")));
  424. illegalShipData.setSystemEsIndex(indexName);
  425. try {
  426. illegalShipDataService.insertIllegalShipData(illegalShipData);
  427. } catch (Exception ignored) {
  428. ignored.printStackTrace();
  429. }
  430. }
  431. }
  432. /**
  433. * 或缺数据库字段的大小写
  434. *
  435. * @param data
  436. * @param <V>
  437. * @return
  438. */
  439. private static <V> LinkedCaseInsensitiveMap<V> toCaseInsensitiveMap(Map<String, V> data) {
  440. LinkedCaseInsensitiveMap map = new LinkedCaseInsensitiveMap();
  441. map.putAll(data);
  442. return map;
  443. }
  444. @Override
  445. public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  446. this.applicationContext = applicationContext;
  447. }
  448. }