增加mongodb相关功能

jeecg-boot-base-core/pom.xml

@@ -214,6 +214,11 @@
 			<groupId>com.fasterxml.jackson.module</groupId>
 			<artifactId>jackson-module-kotlin</artifactId>
 		</dependency>
+		<!-- mongodb  -->
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-mongodb</artifactId>
+		</dependency>
 	</dependencies>
 
 </project>

jeecg-module-system/jeecg-system-start/src/main/resources/application-dev.yml

@@ -194,6 +194,15 @@ spring:
 #    host: 152.136.206.27
 #    port: 6379
 #    password: ''
+
+#  mongodb配置
+  data:
+    mongodb:
+      uri: mongodb://phm:phm123@127.0.0.1:27017/phm
+      print: true
+      slowQuery: true
+      slowTime: 1000
+
 #mybatis plus 设置
 mybatis-plus:
   mapper-locations: classpath*:org/jeecg/modules/**/xml/*Mapper.xml

module_ems/src/main/java/org/jeecg/modules/kafka/consumer/TestConsumer.java

@@ -1,11 +1,24 @@
 package org.jeecg.modules.kafka.consumer;
 
+import net.sf.json.JSONObject;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.jeecg.modules.collectdata.dao.CollectdataRepository;
+import org.jeecg.modules.collectdata.entity.Collectdata;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.stereotype.Component;
 
+import java.util.*;
+import java.text.SimpleDateFormat;
+
 @Component
 public class TestConsumer {
+    @Autowired
+    private MongoTemplate mongoTemplate;
+    @Autowired
+    private CollectdataRepository repository;
 
     /**
      * 指定一个消费者组，一个主题主题。
@@ -13,8 +26,7 @@ public class TestConsumer {
      */
 //    @KafkaListener(topicPattern = "thing___.*.___property")
     @KafkaListener(topicPattern = "#{'${spring.kafka.topic-patterns}'}")
-    public void simpleConsumer(ConsumerRecord<?, ?> record) {
-        System.out.println("进入simpleConsumer方法");
+    public void simpleConsumer(ConsumerRecord<?, ?> record, Acknowledgment ack) {
         System.out.printf("topic2.* = %s, 偏移量 = %d, key = %s, 内容 = %s, 创建消息的时间戳 =%d%n \n",
                 record.topic(),
                 record.offset(),
@@ -22,5 +34,59 @@ public class TestConsumer {
                 record.value(),
                 record.timestamp()
         );
+
+        // 提交（用来标记一条消息已经消费完成,即将从消息队列里移除。）
+        ack.acknowledge();
+        saveDataR(record);
+    }
+
+    /**
+     * 保存数据(通过mongoTemplate保存)
+     * @param record kafka记录
+     */
+    private void saveData(ConsumerRecord<?, ?> record) {
+        JSONObject cd = JSONObject.fromObject(record.value());
+
+        String equipmentcode = record.topic().replace("thing___", "").replace("___property", "");
+        Map<String,Object> map = new HashMap<>();
+        map.put("logid", cd.get("id"));
+        map.put("logtime", record.timestamp());
+        map.put("version", cd.get("version"));
+        map.put("equipmentcode", equipmentcode);
+        map.put("method", cd.get("method"));
+
+        JSONObject jsonObject = (JSONObject) JSONObject.fromObject(record.value()).get("properties");
+        Map<String,Object> map1 = new HashMap<>();
+        for (Object key : jsonObject.keySet()) {
+            map1.put(key.toString(), jsonObject.get(key));
+        }
+        map.put("properties", map1);
+
+        mongoTemplate.insert(map, "tpmcollectdata");
+    }
+
+    /**
+     * 保存数据（通过Repository保存）
+     * @param record kafka记录
+     */
+    private void saveDataR(ConsumerRecord<?, ?> record) {
+        JSONObject cd = JSONObject.fromObject(record.value());
+
+        String equipmentcode = record.topic().replace("thing___", "").replace("___property", "");
+        Collectdata obj = new Collectdata();
+        obj.setLogid(Long.parseLong(cd.get("id").toString()));
+        obj.setLogtime(new Date(record.timestamp()));
+        obj.setVersion(cd.get("version").toString());
+        obj.setEquipmentcode(equipmentcode);
+        obj.setMethod(cd.get("method").toString());
+
+        JSONObject jsonObject = (JSONObject) JSONObject.fromObject(record.value()).get("properties");
+        Map<String,Object> map1 = new HashMap<>();
+        for (Object key : jsonObject.keySet()) {
+            map1.put(key.toString(), jsonObject.get(key));
+        }
+        obj.setProperties(map1);
+
+        repository.save(obj);
     }
 }

module_tpm/src/main/java/org/jeecg/modules/collectdata/controller/CollectdataController.java

@@ -0,0 +1,63 @@
+package org.jeecg.modules.collectdata.controller;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import org.jeecg.modules.collectdata.dao.CollectdataRepository;
+import org.jeecg.modules.collectdata.entity.Collectdata;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.jeecg.common.api.vo.Result;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 测试mongodb
+ */
+@Api(tags="测试mongodb")
+@RestController
+@RequestMapping("/collectdata/collectdata")
+public class CollectdataController {
+    @Autowired
+    private MongoTemplate mongoTemplate;
+    @Autowired
+    private CollectdataRepository repository;
+
+
+
+    @ApiOperation(value="测试mongodb-插入", notes="测试mongodb-插入")
+    @GetMapping("/insert")
+    public Result<?> insert(){
+        Map<String,Object> map = new HashMap<>();
+        map.put("id","4");
+        map.put("version","1.0");
+        map.put("equipmentcode","9010100000009");
+        map.put("method","thing.event.property.post");
+        Map<String,Object> map1 = new HashMap<>();
+        map1.put("time",1645791383383L);
+        map1.put("ALARM", 0);
+        map1.put("ENV_TEMP", 35);
+        map1.put("EVA_OUT_TEMP", 56);
+        map1.put("EVA_IN_TEMP", 10.6);
+        map1.put("COOL_OUT_TEMP", 20);
+        map1.put("COOL_IN_TEMP", 12.5);
+        map.put("properties", map1);
+        mongoTemplate.insert(map, "tpmcollectdata");
+
+        return Result.OK("存入成功");
+    }
+
+    @ApiOperation(value="测试mongodb-查询", notes="测试mongodb-查询")
+    @GetMapping("/querybycode")
+    public Result<?> query(@RequestParam(name="equipmentcode") String equipmentcode) {
+//        List<Collectdata> list = repository.findByEquipmentcode(equipmentcode);
+        List<Collectdata> list = repository.findByEquipmentcode(equipmentcode);
+
+        return Result.OK(list);
+    }
+}

module_tpm/src/main/java/org/jeecg/modules/collectdata/dao/CollectdataRepository.java

@@ -0,0 +1,13 @@
+package org.jeecg.modules.collectdata.dao;
+
+import org.jeecg.modules.collectdata.entity.Collectdata;
+import org.springframework.data.mongodb.repository.MongoRepository;
+
+import java.util.List;
+
+public interface CollectdataRepository extends MongoRepository<Collectdata, String> {
+
+//    Collectdata findByEquipmentcode(String equpmentcod);
+    List<Collectdata> findByEquipmentcode(String equipmentcode);
+
+}

module_tpm/src/main/java/org/jeecg/modules/collectdata/entity/Collectdata.java

@@ -0,0 +1,31 @@
+package org.jeecg.modules.collectdata.entity;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import lombok.Data;
+import org.springframework.data.mongodb.core.mapping.Document;
+import org.springframework.format.annotation.DateTimeFormat;
+
+import javax.persistence.Id;
+import java.util.Date;
+
+@Data
+@Document("tpmcollectdata")
+public class Collectdata {
+
+    @Id
+    public String id;
+
+    public Long logid;
+
+    @JsonFormat(timezone = "GMT+8",pattern = "yyyy-MM-dd HH:mm:ss")
+    @DateTimeFormat(pattern="yyyy-MM-dd HH:mm:ss")
+    public Date logtime;
+
+    public String version;
+
+    public String equipmentcode;
+
+    public String method;
+
+    public Object properties;
+}