kafka ais

ruoyi-admin/src/main/java/com/ruoyi/web/controller/gas/GasCommonController.java

@@ -48,7 +48,6 @@ import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
-import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.Date;
@@ -287,10 +286,6 @@ public class GasCommonController {
         List<IllegalInfo> pageList = Convert.toList(IllegalInfo.class, stringObjectMap.get("pageList"));
         // 查询船舶的实时经纬度
         Map<String, AisShip> aisShipMap = aisShipService.queryDynamicShip(1);
-        List<AisShip> aisShipAddressList = aisShipService.queryShipAddressList(1);
-        // 组装一段时间内的经纬度数据
-        Map<String, List<String>> collect = aisShipAddressList.stream().collect(Collectors.groupingBy(AisShip::getMmsi, Collectors.mapping(a -> a.getLng() + "," + a.getLat(), Collectors.toList())));
-        Map<String, List<BigDecimal>> headCollect = aisShipAddressList.stream().collect(Collectors.groupingBy(AisShip::getMmsi, Collectors.mapping(AisShip::getHead, Collectors.toList())));
         for (IllegalInfo illegalInfo : pageList) {
             if (StrUtil.isNotBlank(illegalInfo.getMmsi())) {
                 AisShip aisShip = aisShipMap.get(illegalInfo.getMmsi());
@@ -302,18 +297,6 @@ public class GasCommonController {
                     illegalInfo.setSpeed(ObjectUtil.isEmpty(aisShip.getSpeed()) ? "-" : aisShip.getSpeed() + "节");
                     illegalInfo.setCourse(ObjectUtil.isEmpty(aisShip.getCourse()) ? "-" : aisShip.getCourse() + "");
                 }
-                // 数组转二维数组
-                List<String> lnglat = collect.get(illegalInfo.getMmsi());
-                Double[][] covert = new Double[lnglat.size()][];
-                for (int i = 0; i < lnglat.size(); i++) {
-                    String[] second = lnglat.get(i).split(",");
-                    covert[i] = new Double[second.length];
-                    for (int j = 0; j < second.length; j++) {
-                        covert[i][j] = Double.parseDouble(second[j]);
-                    }
-                }
-                illegalInfo.setLnglat(covert);
-                illegalInfo.setHeads(headCollect.get(illegalInfo.getMmsi()));
             }
         }
         pageList =

ruoyi-admin/src/main/java/com/ruoyi/web/core/config/AisConsumer.java

@@ -0,0 +1,27 @@
+package com.ruoyi.web.core.config;
+
+import com.ruoyi.web.service.IGasCommonService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Description: TODO
+ * @Author: huangcheng
+ * @Date: 2021/11/30
+ * @Version V1.0
+ */
+@Component
+@Profile({"prod1"})
+public class AisConsumer {
+
+    @Autowired
+    private IGasCommonService gasCommonService;
+
+    @KafkaListener(topics = {"ais-merge"})
+    public void receive(String message) {
+        gasCommonService.dealAis(message);
+    }
+
+}

ruoyi-admin/src/main/java/com/ruoyi/web/service/IGasCommonService.java

@@ -10,4 +10,6 @@ import com.ruoyi.system.domain.SyncPost;
  */
 public interface IGasCommonService {
     void sync(SyncPost syncPost);
+
+    void dealAis(String message);
 }

ruoyi-admin/src/main/java/com/ruoyi/web/service/Impl/GasCommonServiceImpl.java

@@ -1,15 +1,19 @@
 package com.ruoyi.web.service.Impl;
 
 import cn.hutool.core.convert.Convert;
+import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.util.NumberUtil;
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.http.HttpUtil;
+import cn.hutool.json.JSONUtil;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.ruoyi.common.constant.ElasticConstants;
+import com.ruoyi.common.utils.uuid.IdUtils;
 import com.ruoyi.framework.config.ElasticSearchClient;
+import com.ruoyi.system.domain.HuiHAisInfo;
 import com.ruoyi.system.domain.SyncPost;
 import com.ruoyi.system.domain.vo.AisShipInfo;
 import com.ruoyi.system.mapper.GasCommonMapper;
@@ -193,4 +197,13 @@ public class GasCommonServiceImpl implements IGasCommonService {
         }
     }
 
+    @SneakyThrows
+    @Override
+    @Async
+    public void dealAis(String message) {
+        HuiHAisInfo aisBaseInfo = JSONUtil.toBean(message, HuiHAisInfo.class);
+        aisBaseInfo.setId(IdUtils.fastSimpleUUID());
+        aisBaseInfo.setCreateTime(DateUtil.date());
+        gasCommonMapper.saveHuiHAisInfo(aisBaseInfo);
+    }
 }

ruoyi-admin/src/main/resources/application.yml

@@ -56,7 +56,18 @@ spring:
     restart:
       # 热部署开关
       enabled: true
-
+  # kafka配置
+  # 指定消息被消费之后自动提交偏移量
+  # （即消息的编号，表示消费到了哪个位置，
+  # 消费者每消费完一条消息就会向kafka服务器汇报自己消消费到的那个消息的编号，以便于下次继续消费）
+  kafka:
+    consumer:
+      enable-auto-commit: true
+      group-id: xintong_cbwq
+      auto-offset-reset: latest
+      bootstrap-servers: 198.17.188.71:9092
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
 
 # token配置
 token:
@@ -91,4 +102,4 @@ xss:
   # 排除链接（多个用逗号分隔）
   excludes: /system/notice/*
   # 匹配链接
-  urlPatterns: /system/*,/monitor/*,/tool/*
+  urlPatterns: /system/*,/monitor/*,/tool/*

ruoyi-framework/pom.xml

@@ -18,7 +18,7 @@
     <dependencies>
 
         <!-- SpringBoot Web容器 -->
-         <dependency>
+        <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
@@ -116,6 +116,11 @@
             <version>1.1.5</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+
     </dependencies>
 
 </project>

ruoyi-system/src/main/java/com/ruoyi/system/domain/HuiHAisInfo.java

@@ -0,0 +1,40 @@
+package com.ruoyi.system.domain;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import lombok.Data;
+
+import java.util.Date;
+
+/**
+ * @Description: TODO
+ * @Author: huangcheng
+ * @Date: 2021/11/30
+ * @Version V1.0
+ */
+@Data
+public class HuiHAisInfo {
+    private String id;
+    private String ship_no;
+    private String mmsi;
+    private String speed;
+    private String head;
+    private String rate;
+    private String course;
+    private String tel;
+    private String power;
+    private String lat;
+    private String ship_id;
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    private Date receive;
+    private String lng;
+    private String length;
+    private String shipname_cn;
+    private String ship_type;
+    private String ship_reg_no;
+    private String depth;
+    private String draught;
+    private String width;
+    private String ship_type_code;
+    private String shipname;
+    private Date createTime;
+}

ruoyi-system/src/main/java/com/ruoyi/system/domain/IllegalInfo.java

@@ -5,7 +5,6 @@ import lombok.Data;
 
 import java.math.BigDecimal;
 import java.util.Date;
-import java.util.List;
 
 /**
  * 嫌疑船舶
@@ -41,10 +40,6 @@ public class IllegalInfo {
     // 船舶实时经纬度
     private BigDecimal lat;
 
-    private Double[][] lnglat;
-
-    private List<BigDecimal> heads;
-
     // 船舶实时经纬度上传时间
     @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
     private Date receive;

ruoyi-system/src/main/java/com/ruoyi/system/mapper/GasCommonMapper.java

@@ -2,6 +2,7 @@ package com.ruoyi.system.mapper;
 
 import com.ruoyi.common.annotation.DataSource;
 import com.ruoyi.common.enums.DataSourceType;
+import com.ruoyi.system.domain.HuiHAisInfo;
 import com.ruoyi.system.domain.SyncPost;
 import org.apache.ibatis.annotations.Param;
 
@@ -17,7 +18,7 @@ import java.util.Map;
 public interface GasCommonMapper {
 
     @DataSource(DataSourceType.SLAVE_AIS)
-    List<Map<String,Object>> querySemInstrument(@Param("table") String table);
+    List<Map<String, Object>> querySemInstrument(@Param("table") String table);
 
     @DataSource(DataSourceType.SLAVE_AIS)
     List<Map<String, Object>> queryIllegalShip(@Param("syncPost") SyncPost syncPost);
@@ -33,4 +34,6 @@ public interface GasCommonMapper {
 
     @DataSource(DataSourceType.SLAVE_SO2)
     List<Map<String, Object>> queryShipStaticInfo(@Param("syncPost") SyncPost syncPost);
+
+    void saveHuiHAisInfo(HuiHAisInfo aisBaseInfo);
 }

ruoyi-system/src/main/resources/mapper/system/GasCommonMapper.xml

@@ -3,6 +3,59 @@
         PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.ruoyi.system.mapper.GasCommonMapper">
+    <insert id="saveHuiHAisInfo" parameterType="huiHAisInfo">
+        insert into ais_ship_info
+        <trim prefix="(" suffix=")" suffixOverrides=",">
+            <if test="id != null">id,</if>
+            <if test="ship_no != null">ship_no,</if>
+            <if test="mmsi != null">mmsi,</if>
+            <if test="speed != null">speed,</if>
+            <if test="head != null">head,</if>
+            <if test="rate != null">rate,</if>
+            <if test="course != null">course,</if>
+            <if test="tel != null">tel,</if>
+            <if test="power != null">power,</if>
+            <if test="lat != null">lat,</if>
+            <if test="lng != null">lng,</if>
+            <if test="ship_id != null">ship_id,</if>
+            <if test="receive != null">receive,</if>
+            <if test="length != null">length,</if>
+            <if test="shipname_cn != null">shipname_cn,</if>
+            <if test="ship_type != null">ship_type,</if>
+            <if test="ship_reg_no != null">ship_reg_no,</if>
+            <if test="depth != null">depth,</if>
+            <if test="draught != null">draught,</if>
+            <if test="width != null">width,</if>
+            <if test="ship_type_code != null">ship_type_code,</if>
+            <if test="shipname != null">shipname,</if>
+            <if test="createTime != null">create_time,</if>
+        </trim>
+        <trim prefix="values (" suffix=")" suffixOverrides=",">
+            <if test="id != null">#{id},</if>
+            <if test="ship_no != null">#{ship_no},</if>
+            <if test="mmsi != null">#{mmsi},</if>
+            <if test="speed != null">#{speed},</if>
+            <if test="head != null">#{head},</if>
+            <if test="rate != null">#{rate},</if>
+            <if test="course != null">#{course},</if>
+            <if test="tel != null">#{tel},</if>
+            <if test="power != null">#{power},</if>
+            <if test="lat != null">#{lat},</if>
+            <if test="lng != null">#{lng},</if>
+            <if test="ship_id != null">#{ship_id},</if>
+            <if test="receive != null">#{receive},</if>
+            <if test="length != null">#{length},</if>
+            <if test="shipname_cn != null">#{shipname_cn},</if>
+            <if test="ship_type != null">#{ship_type},</if>
+            <if test="ship_reg_no != null">#{ship_reg_no},</if>
+            <if test="depth != null">#{depth},</if>
+            <if test="draught != null">#{draught},</if>
+            <if test="width != null">#{width},</if>
+            <if test="ship_type_code != null">#{ship_type_code},</if>
+            <if test="shipname != null">#{shipname},</if>
+            <if test="createTime != null">#{createTime},</if>
+        </trim>
+    </insert>
 
     <select id="querySemInstrument" resultType="java.util.Map">
         select id,