批量写入

price_collection/src/main/java/com/liangjiang/price_collection/dto/BackupsInfo.java

@@ -2,9 +2,15 @@ package com.liangjiang.price_collection.dto;
 
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
 @Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
 @TableName("info_backups")
 public class BackupsInfo {
     // 时间

price_collection/src/main/java/com/liangjiang/price_collection/dto/PriceInfoDto.java

@@ -16,4 +16,8 @@ public class PriceInfoDto {
     private String bid;
     // 卖价
     private String ask;
+
+    public BackupsInfo convertAdd(){
+        return BackupsInfo.builder().time(this.getTime()).ask(this.getAsk()).bid(this.getBid()).tableName(String.format("%s_%s_%s", this.getCoin().replaceAll("_", ""), this.getSource(), this.getB_type()).toUpperCase()).build();
+    }
 }

price_collection/src/main/java/com/liangjiang/price_collection/mapper/TableMapper.java

@@ -1,5 +1,6 @@
 package com.liangjiang.price_collection.mapper;
 
+import com.liangjiang.price_collection.dto.BackupsInfo;
 import com.liangjiang.price_collection.dto.PriceDto;
 import com.liangjiang.price_collection.dto.PriceInfoDto;
 import org.apache.ibatis.annotations.Mapper;
@@ -7,14 +8,19 @@ import org.apache.ibatis.annotations.Param;
 
 import java.sql.SQLSyntaxErrorException;
 import java.util.List;
+import java.util.Set;
 
 @Mapper
 public interface TableMapper {
     // 创建表
     void createTable(@Param("tableName")String tableName);
+    // 批量创建表
+    void createTables(@Param("tableNames")Set<String> tableNames);
     // 获取所有表名
     List<String> getTableName();
 
+    void savePrices(@Param("priceInfos") List<BackupsInfo> priceInfos);
+
     // 保存价格信息
     void savePrice(@Param("tableName") String tableName, @Param("id") Integer id, @Param("ask") String ask,@Param("bid") String bid) throws SQLSyntaxErrorException;
 

price_collection/src/main/java/com/liangjiang/price_collection/service/impl/QueryService.java

@@ -39,10 +39,12 @@ public class QueryService {
         // 收集查询结果
         for (Future<ExchangeCoinPriceDto> future : futures) {
             ExchangeCoinPriceDto dto = future.get();
-            results.add(dto);
+            if(!dto.getPrices().isEmpty()){
+                results.add(dto);
+            }
         }
         if(CollectionUtil.isEmpty(results)){
-            return null;
+            return PriceListDto.builder().time(new LinkedList<>()).values(new LinkedList<>()).build();
         }
         List<Future<ExchangeCoinPriceArrDto>> resultFutures = new LinkedList<>();
         List<ExchangeCoinPriceArrDto> data = new LinkedList<>();

price_collection/src/main/java/com/liangjiang/price_collection/service/impl/TableServiceImpl.java

@@ -88,9 +88,29 @@ public class TableServiceImpl implements ITableService {
 
     @Override
     public void savePriceBatch(List<PriceInfoDto> dtos){
-        dtos.forEach(this::savePrice);
+        if(CollectionUtil.isEmpty(dtos)){
+            return;
+        }
+        Set<String> tableNames = new HashSet<>();
+        List<BackupsInfo> addDtos = new LinkedList<>();
+        dtos.forEach(dto -> {
+            String tableName = String.format("%s_%s_%s", dto.getCoin().replaceAll("_", ""), dto.getSource(), dto.getB_type()).toUpperCase();
+            tableNames.add(tableName);
+            addDtos.add(dto.convertAdd());
+        });
+        try {
+            // 建表
+            this.tableMapper.createTables(tableNames);
+            // 写入数据
+            this.tableMapper.savePrices(addDtos);
+        } catch (Exception ex){
+            log.error("插入报错", ex);
+            backupsInfoService.saveBatch(addDtos);
+        }
     }
 
+
+
     @Override
     public void savePrice(PriceInfoDto dto){
         String tableName = String.format("%s_%s_%s", dto.getCoin().replaceAll("_", ""), dto.getSource(), dto.getB_type()).toUpperCase();

price_collection/src/main/resources/application.yml

@@ -31,13 +31,13 @@ spring:
       testOnBorrow: false
       testOnReturn: false
       poolPreparedStatements: true
-      # 配置监控统计拦截的filters，去掉后监控界面sql无法统计。‘wall’用于防火墙
-      filters: stat,wall
       filter:
         wall:
           config:
             none-base-statement-allow: true
             multi-statement-allow: true
+      # 配置监控统计拦截的filters，去掉后监控界面sql无法统计。‘wall’用于防火墙
+#      filters: stat,wall
 
 #  redis:
 #    # Redis服务器地址

price_collection/src/main/resources/mapper/TableMapper.xml

@@ -11,6 +11,17 @@
         )
     </update>
 
+    <update id="createTables" parameterType="java.util.Set">
+        <foreach item="item" collection="tableNames" open="" separator=";" close="">
+            CREATE TABLE IF NOT EXISTS ${item} (
+            `id` INT UNSIGNED NOT NULL,
+            `bid` VARCHAR(128) NOT NULL,
+            `ask` VARCHAR(128) NOT NULL,
+            PRIMARY KEY (`id`)
+            )
+        </foreach>
+    </update>
+
     <!-- 获取所有数据表格 -->
     <select id="getTableName" resultType="java.lang.String">
         SELECT
@@ -25,6 +36,12 @@
             table_name
     </select>
 
+    <insert id="savePrices">
+        <foreach item="item" collection="priceInfos" open="" separator=";" close="">
+            INSERT ${item.tableName} ( id, bid, ask ) VALUE (${item.time}, ${item.bid}, ${item.ask})on duplicate key update bid = ${item.bid}, ask = ${item.ask}
+        </foreach>
+    </insert>
+
     <insert id="savePrice">
         INSERT ${tableName} ( id, bid, ask ) VALUE (${id}, ${bid}, ${ask})on duplicate key update bid = ${bid}, ask = ${ask}
     </insert>