查询接口添加

1. 查询接口添加
2. 跨域解决
3. 完善数据添加

price_collection/src/main/java/com/liangjiang/price_collection/controller/ApiController.java

@@ -1,21 +1,75 @@
 package com.liangjiang.price_collection.controller;
 
+import com.liangjiang.price_collection.dto.ExchangeCoinPriceArrDto;
 import com.liangjiang.price_collection.dto.PriceInfoDto;
 import com.liangjiang.price_collection.service.ITableService;
+import com.liangjiang.price_collection.service.impl.QueryService;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.web.bind.annotation.*;
 
+import java.util.LinkedList;
 import java.util.List;
+import java.util.Set;
 
-@CrossOrigin
+@Slf4j
 @RestController
 @RequiredArgsConstructor
 @RequestMapping("/api")
 public class ApiController {
     private final ITableService tableService;
+    private final QueryService queryService;
 
     @PostMapping("/add")
     public void saveInfo(@RequestBody List<PriceInfoDto> priceInfoDtos){
         tableService.savePriceBatch(priceInfoDtos);
     }
+
+    @GetMapping("/getExchange")
+    public Set<String> exchangeSelect(){
+       return tableService.getExchangeSelect();
+    }
+
+    @GetMapping("/getCoin")
+    public Set<String> coinSelect(@RequestParam("exchange") String exchange){
+        String[] params = exchange.split(",");
+        return tableService.getCoinSelect(params);
+    }
+
+    @GetMapping("/getPrices")
+    public List<ExchangeCoinPriceArrDto> getPriceRest(@RequestParam("exchange") String exchange, @RequestParam("coin")String coin){
+        Integer[] times = QueryService.getDaysAgoTime(System.currentTimeMillis(), 3L);
+        try {
+            List<ExchangeCoinPriceArrDto> a =  queryService.performQueries(exchange.split(","), coin, times[1], times[0]);
+            return a;
+        } catch (Exception ex){
+            log.error("查询出错", ex);
+            return new LinkedList<>();
+        }
+    }
+
+    @GetMapping("/getPricesRest")
+    public List<ExchangeCoinPriceArrDto> getPrices(@RequestParam("exchange") String exchange, @RequestParam("coin")String coin){
+        System.out.println("exchange = " + exchange + ", coin = " + coin);
+        List<ExchangeCoinPriceArrDto> result = new LinkedList<>();
+        ExchangeCoinPriceArrDto dto1 = ExchangeCoinPriceArrDto.builder().exchange("BINANCE_SPOT").symbol("BTCUSDT").ask(new LinkedList<>()).bid(new LinkedList<>()).time(new LinkedList<>()).build();
+        ExchangeCoinPriceArrDto dto2 = ExchangeCoinPriceArrDto.builder().exchange("GATE_SPOT").symbol("BTCUSDT").ask(new LinkedList<>()).bid(new LinkedList<>()).time(new LinkedList<>()).build();
+        int time = 1565613908;
+        for (int i = 1; i < 5; i ++){
+            int x = i * 2 + 1;
+            int y = i * 2 - 2;
+            dto1.getAsk().add(Integer.toString(x));
+            dto1.getBid().add(Integer.toString(y));
+            dto1.getTime().add(time);
+            int x1 = i * 2;
+            int y1 = i * 2 - 1;
+            dto2.getAsk().add(Integer.toString(x1));
+            dto2.getBid().add(Integer.toString(y1));
+            dto2.getTime().add(time);
+            time ++;
+        }
+        result.add(dto1);
+        result.add(dto2);
+        return result;
+    }
 }

price_collection/src/main/java/com/liangjiang/price_collection/controller/TestController.java

@@ -1,26 +1,25 @@
 package com.liangjiang.price_collection.controller;
-
-import com.liangjiang.price_collection.dto.PriceInfoDto;
-import com.liangjiang.price_collection.service.ITableService;
-import lombok.RequiredArgsConstructor;
-import org.springframework.web.bind.annotation.*;
-
-import java.util.List;
-
-@CrossOrigin
-@RestController
-@RequiredArgsConstructor
-@RequestMapping("/test")
+//
+//import com.liangjiang.price_collection.dto.PriceInfoDto;
+//import com.liangjiang.price_collection.service.ITableService;
+//import lombok.RequiredArgsConstructor;
+//import org.springframework.web.bind.annotation.*;
+//
+//import java.util.List;
+//
+//@RestController
+//@RequiredArgsConstructor
+//@RequestMapping("/test")
 public class TestController {
-    private final ITableService tableService;
-
-    @RequestMapping(value = "/createTable", method = RequestMethod.POST)
-    public Boolean createTable(@RequestBody PriceInfoDto priceInfoDto) {
-        return this.tableService.createTable(priceInfoDto.getSource(), priceInfoDto.getCoin(), priceInfoDto.getType());
-    }
-
-    @RequestMapping(value = "/getTableName", method = RequestMethod.GET)
-    public List<String> getTableName() {
-        return this.tableService.getTableNames();
-    }
+//    private final ITableService tableService;
+//
+//    @RequestMapping(value = "/createTable", method = RequestMethod.POST)
+//    public Boolean createTable(@RequestBody PriceInfoDto priceInfoDto) {
+//        return this.tableService.createTable(priceInfoDto.getSource(), priceInfoDto.getCoin(), priceInfoDto.getType());
+//    }
+//
+//    @RequestMapping(value = "/getTableName", method = RequestMethod.GET)
+//    public List<String> getTableName() {
+//        return this.tableService.getTableNames();
+//    }
 }

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

@@ -0,0 +1,16 @@
+package com.liangjiang.price_collection.dto;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+@Builder
+public class ExchangeCoinPriceArrDto {
+    private String exchange;
+    private String symbol;
+    private List<Integer> time;
+    private List<String> ask;
+    private List<String> bid;
+}

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

@@ -0,0 +1,13 @@
+package com.liangjiang.price_collection.dto;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.util.TreeMap;
+
+@Data
+@Builder
+public class ExchangeCoinPriceDto {
+    private String exchange;
+    private TreeMap<Integer, PriceDto> prices;
+}

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

@@ -0,0 +1,13 @@
+package com.liangjiang.price_collection.dto;
+
+import lombok.Data;
+
+@Data
+public class PriceDto {
+    // 时间戳 秒级
+    private Integer time;
+    // 买价
+    private String bid;
+    // 卖价
+    private String ask;
+}

price_collection/src/main/java/com/liangjiang/price_collection/filter/CoreFilter.java

@@ -0,0 +1,50 @@
+package com.liangjiang.price_collection.filter;
+
+import org.springframework.core.annotation.Order;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Component;
+
+import javax.servlet.*;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+
+@Order(1)
+@Component
+public class CoreFilter implements Filter {
+
+    @Override
+    public void destroy() {
+
+    }
+
+    @Override
+    public void doFilter(ServletRequest request, ServletResponse response,
+                         FilterChain chain) throws IOException, ServletException {
+        HttpServletRequest req = (HttpServletRequest) request;
+        HttpServletResponse rep = (HttpServletResponse) response;
+
+        //设置允许跨域的配置
+        // 这里填写你允许进行跨域的主机ip（正式上线时可以动态配置具体允许的域名和IP）
+        rep.setHeader("Access-Control-Allow-Origin", "*");
+        // 允许的访问方法
+        rep.setHeader("Access-Control-Allow-Methods", "POST, GET, OPTIONS");
+        // Access-Control-Max-Age 用于 CORS 相关配置的缓存
+        rep.setHeader("Access-Control-Max-Age", "3600");
+        // 允许前端发送的请求头
+        rep.setHeader("Access-Control-Allow-Headers", "auth, Origin, X-Requested-With, Content-Type, Accept");
+        //若要返回cookie、携带seesion等信息则将此项设置我true
+        rep.setHeader("Access-Control-Allow-Credentials", "true");
+        if ("OPTIONS".equals(req.getMethod())){
+            rep.setStatus(HttpStatus.ACCEPTED.value());
+            return;
+        }
+        chain.doFilter(req, rep);
+
+    }
+
+    @Override
+    public void init(FilterConfig arg0) {
+
+    }
+}

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.PriceDto;
 import com.liangjiang.price_collection.dto.PriceInfoDto;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
@@ -18,5 +19,5 @@ public interface TableMapper {
     void savePrice(@Param("tableName") String tableName, @Param("id") Integer id, @Param("ask") String ask,@Param("bid") String bid) throws SQLSyntaxErrorException;
 
     // 获取价格列表
-    List<PriceInfoDto> getPriceInfo(@Param("tableName") String tableName,@Param("endTime") Integer endTime);
+    List<PriceDto> getPriceInfo(@Param("tableName") String tableName, @Param("startTime") Integer startTime, @Param("endTime") Integer endTime);
 }

price_collection/src/main/java/com/liangjiang/price_collection/service/ITableService.java

@@ -1,8 +1,10 @@
 package com.liangjiang.price_collection.service;
 
+import com.liangjiang.price_collection.dto.PriceDto;
 import com.liangjiang.price_collection.dto.PriceInfoDto;
 
 import java.util.List;
+import java.util.Set;
 
 public interface ITableService {
     // 创建表
@@ -16,4 +18,13 @@ public interface ITableService {
 
     // 批量写入
     void savePriceBatch(List<PriceInfoDto> dtos);
+
+    // 获取已有交易所
+    Set<String> getExchangeSelect();
+
+    // 获取交集币对
+    Set<String> getCoinSelect(String[] exchanges);
+
+    // 获取价格信息
+    List<PriceDto> getPriceInfo(String tableName, Integer startTime, Integer endTime);
 }

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

@@ -0,0 +1,92 @@
+package com.liangjiang.price_collection.service.impl;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.liangjiang.price_collection.dto.ExchangeCoinPriceArrDto;
+import com.liangjiang.price_collection.dto.ExchangeCoinPriceDto;
+import com.liangjiang.price_collection.dto.PriceDto;
+import com.liangjiang.price_collection.service.ITableService;
+import com.liangjiang.price_collection.utils.DataAssembleTask;
+import com.liangjiang.price_collection.utils.TableQueryTask;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+@Service
+public class QueryService {
+    private final ExecutorService executorService;
+    private final ITableService tableService;
+
+    @Autowired
+    public QueryService(ITableService tableService) {
+        this.executorService = Executors.newFixedThreadPool(5); // 根据需要调整线程池大小
+        this.tableService = tableService;
+    }
+
+    public List<ExchangeCoinPriceArrDto> performQueries(String[] exchanges, String coin, Integer startTime, Integer endTime) throws Exception {
+        List<Future<ExchangeCoinPriceDto>> futures = new ArrayList<>();
+        Set<ExchangeCoinPriceDto> results = new HashSet<>();
+        // 分发查询
+        for (String exchange : exchanges) {
+            TableQueryTask queryTask = new TableQueryTask(tableService, coin, exchange, startTime, endTime);
+            futures.add(executorService.submit(queryTask));
+        }
+        // 收集查询结果
+        for (Future<ExchangeCoinPriceDto> future : futures) {
+            ExchangeCoinPriceDto dto = future.get();
+            results.add(dto);
+        }
+        if(CollectionUtil.isEmpty(results)){
+            return null;
+        }
+        List<Future<ExchangeCoinPriceArrDto>> resultFutures = new LinkedList<>();
+        List<ExchangeCoinPriceArrDto> data = new LinkedList<>();
+        // 获取时间段的所有秒级
+        List<Integer> times = generateArithmeticSequence(startTime, endTime);
+
+        for (ExchangeCoinPriceDto dto : results){
+            DataAssembleTask task = new DataAssembleTask(dto, coin, times);
+            resultFutures.add(executorService.submit(task));
+        }
+
+        // 收集格式化结果
+        for (Future<ExchangeCoinPriceArrDto> future : resultFutures) {
+            ExchangeCoinPriceArrDto dto = future.get();
+            data.add(dto);
+        }
+        return data;
+    }
+
+    // 用于优雅关闭服务
+    public void shutdown() {
+        executorService.shutdown();
+    }
+
+    public static Integer[] getDaysAgoTime(long startTime, long dayNum){
+        int startTimeInt = Math.toIntExact(startTime / 1000);
+        // 计算3天的毫秒数 (3天 * 24小时 * 60分钟 * 60秒 * 1000毫秒)
+        long threeDaysInMillis = dayNum * 24 * 60 * 60 * 1000;
+        // 计算3天前的毫秒级时间戳
+        long threeDaysAgoMillis = startTime - threeDaysInMillis;
+        // 将毫秒级时间戳转换为秒级时间戳
+        long threeDaysAgoSeconds = threeDaysAgoMillis / 1000;
+        return new Integer[]{startTimeInt, Math.toIntExact(threeDaysAgoSeconds)};
+    }
+
+    public static List<Integer> generateArithmeticSequence(int start, int end) {
+        if (start > end) {
+            System.out.println("start = " + start + ", end = " + end);
+            throw new IllegalArgumentException("首项必须小于或等于末项");
+        }
+
+        List<Integer> sequence = new ArrayList<>();
+        for (int i = start; i <= end; i++) {
+            sequence.add(i);
+        }
+        return sequence;
+    }
+}

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

@@ -1,8 +1,10 @@
 package com.liangjiang.price_collection.service.impl;
 
+import cn.hutool.core.collection.CollectionUtil;
 import com.google.common.collect.Interner;
 import com.google.common.collect.Interners;
 import com.liangjiang.price_collection.dto.BackupsInfo;
+import com.liangjiang.price_collection.dto.PriceDto;
 import com.liangjiang.price_collection.dto.PriceInfoDto;
 import com.liangjiang.price_collection.mapper.TableMapper;
 import com.liangjiang.price_collection.service.IBackupsInfoService;
@@ -12,7 +14,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.jdbc.BadSqlGrammarException;
 import org.springframework.stereotype.Service;
 
-import java.util.List;
+import java.util.*;
 
 @Slf4j
 @Service
@@ -36,6 +38,49 @@ public class TableServiceImpl implements ITableService {
         }
     }
 
+    @Override
+    public Set<String> getExchangeSelect (){
+        List<String> tableNames = this.getTableNames();
+        if (CollectionUtil.isEmpty(tableNames)){
+            return new HashSet<>();
+        }
+        Set<String> exchanges = new HashSet<>();
+
+        for (String tableName : tableNames){
+            String[] infos = tableName.split("_");
+            // 交易所和类型
+            exchanges.add(String.format("%s_%s", infos[1], infos[2]));
+        }
+        return exchanges;
+    }
+
+    @Override
+    public Set<String> getCoinSelect(String[] exchanges){
+        if(exchanges == null || exchanges.length == 0){
+            return new HashSet<>();
+        }
+        List<String> tableNames = this.getTableNames();
+        if (CollectionUtil.isEmpty(tableNames)){
+            return new HashSet<>();
+        }
+        Map<String, Set<String>> exchange_coin = new HashMap<>();
+        for (String exchange : exchanges){
+            exchange_coin.put(exchange, new HashSet<>());
+        }
+        for (String tableName : tableNames){
+            String[] infos = tableName.split("_");
+            // 交易所和类型
+            String key = String.format("%s_%s", infos[1], infos[2]);
+            if (exchange_coin.containsKey(key)){
+                // 存在加入币对
+                exchange_coin.get(key).add(infos[0]);
+            }
+        }
+        Set<Set<String>> values = new HashSet<>(exchange_coin.values());
+        // 返回交集
+        return getUnion(values);
+    }
+
     @Override
     public List<String> getTableNames(){
         return tableMapper.getTableName();
@@ -77,4 +122,17 @@ public class TableServiceImpl implements ITableService {
             backupsInfoService.save(backupsInfo);
         }
     }
+
+    @Override
+    public List<PriceDto> getPriceInfo(String tableName, Integer startTime, Integer endTime){
+        return this.tableMapper.getPriceInfo(tableName, startTime, endTime);
+    }
+
+    public static Set<String> getUnion(Set<Set<String>> sets) {
+        Set<String> result = new HashSet<>();
+        for (Set<String> set : sets) {
+            result.addAll(set); // 将每个集合的元素添加到结果集合中
+        }
+        return result;
+    }
 }

price_collection/src/main/java/com/liangjiang/price_collection/utils/DataAssembleTask.java

@@ -0,0 +1,51 @@
+package com.liangjiang.price_collection.utils;
+
+import com.liangjiang.price_collection.dto.ExchangeCoinPriceArrDto;
+import com.liangjiang.price_collection.dto.ExchangeCoinPriceDto;
+import com.liangjiang.price_collection.dto.PriceDto;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.TreeMap;
+import java.util.concurrent.Callable;
+
+@Data
+@AllArgsConstructor
+public class DataAssembleTask  implements Callable<ExchangeCoinPriceArrDto> {
+    private ExchangeCoinPriceDto dto;
+    private String coin;
+    private List<Integer> times;
+
+    public ExchangeCoinPriceArrDto call() {
+        ExchangeCoinPriceArrDto exchangeCoinPrice = ExchangeCoinPriceArrDto.builder()
+                .exchange(dto.getExchange())
+                .symbol(coin)
+                .time(new LinkedList<>())
+                .ask(new LinkedList<>()).bid(new LinkedList<>()).build();
+        for(Integer time:times){
+            TreeMap<Integer, PriceDto> priceDtoTreeMap = dto.getPrices();
+            String bid;
+            String ask;
+            if (priceDtoTreeMap.isEmpty()){ // 为空的情况
+                bid = "0";
+                ask = "0";
+            } else {
+                // 获取最近的key，<= time
+                Integer key = dto.getPrices().floorKey(time);
+                if(key == null){
+                    key = dto.getPrices().ceilingKey(time);
+                }
+                PriceDto priceDto = dto.getPrices().get(key);
+                bid = priceDto.getBid();
+                ask = priceDto.getAsk();
+            }
+            exchangeCoinPrice.getAsk().add(ask);
+            exchangeCoinPrice.getBid().add(bid);
+            exchangeCoinPrice.getTime().add(time);
+        }
+        return exchangeCoinPrice;
+    }
+}

price_collection/src/main/java/com/liangjiang/price_collection/utils/TableQueryTask.java

@@ -0,0 +1,37 @@
+package com.liangjiang.price_collection.utils;
+
+import com.liangjiang.price_collection.dto.ExchangeCoinPriceDto;
+import com.liangjiang.price_collection.dto.PriceDto;
+import com.liangjiang.price_collection.service.ITableService;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+import java.util.TreeMap;
+import java.util.concurrent.Callable;
+import java.util.stream.Collectors;
+
+@Data
+@Slf4j
+@AllArgsConstructor
+public class TableQueryTask implements Callable<ExchangeCoinPriceDto> {
+    private ITableService tableService;
+    private String coin;
+    private String exchange;
+    private Integer startTime;
+    private Integer endTime;
+
+    @Override
+    public ExchangeCoinPriceDto call() {
+        try {
+            String tableName = String.format("%s_%s", this.coin.toUpperCase(), this.exchange.toUpperCase());
+            List<PriceDto> price = tableService.getPriceInfo(tableName, this.startTime, this.endTime);
+            TreeMap<Integer, PriceDto> timePrice = price.stream().collect(Collectors.toMap(PriceDto::getTime, p -> p, (o, n) -> n, TreeMap::new));
+            return ExchangeCoinPriceDto.builder().exchange(exchange).prices(timePrice).build();
+        }catch (Exception ex){
+            log.error("查询错误", ex);
+            return ExchangeCoinPriceDto.builder().exchange(exchange).prices(new TreeMap<>()).build();
+        }
+    }
+}

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

@@ -20,6 +20,7 @@
         WHERE
             table_type = 'BASE TABLE'
           AND table_schema = 'one_price_warehouse'
+          and table_name != 'info_backups'
         ORDER BY
             table_name
     </select>
@@ -28,7 +29,7 @@
         INSERT ${tableName} ( id, bid, ask ) VALUE (${id}, ${bid}, ${ask})on duplicate key update bid = ${bid}, ask = ${ask}
     </insert>
 
-    <select id="getPriceInfo" resultType="com.liangjiang.price_collection.dto.PriceInfoDto">
-        select id, bid, ask from ${tableName} where id > ${timeEnd}
+    <select id="getPriceInfo" resultType="com.liangjiang.price_collection.dto.PriceDto">
+        select id as `time`, bid, ask from ${tableName} where id between ${startTime} and ${endTime}
     </select>
 </mapper>

price_collection/target/classes/mapper/TableMapper.xml

@@ -20,6 +20,7 @@
         WHERE
             table_type = 'BASE TABLE'
           AND table_schema = 'one_price_warehouse'
+          and table_name != 'info_backups'
         ORDER BY
             table_name
     </select>
@@ -28,7 +29,7 @@
         INSERT ${tableName} ( id, bid, ask ) VALUE (${id}, ${bid}, ${ask})on duplicate key update bid = ${bid}, ask = ${ask}
     </insert>
 
-    <select id="getPriceInfo" resultType="com.liangjiang.price_collection.dto.PriceInfoDto">
-        select id, bid, ask from ${tableName} where id > ${timeEnd}
+    <select id="getPriceInfo" resultType="com.liangjiang.price_collection.dto.PriceDto">
+        select id as `time`, bid, ask from ${tableName} where id between ${startTime} and ${endTime}
     </select>
 </mapper>