blame | history | patch | commit | commitdiff | ignore whitespace
1
zj
2024-07-21 5c1e682edc2bc6c89cf0f34f93a438d1da274e64 
 websocketClient/src/main/java/org/example/WsBean/MexcWsBean.java


@@ -1,6 +1,8 @@


package org.example.WsBean;




import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;


import com.fasterxml.jackson.core.JsonProcessingException;


import com.fasterxml.jackson.databind.ObjectMapper;


import com.google.common.reflect.TypeToken;


import com.google.gson.Gson;


import lombok.extern.slf4j.Slf4j;


@@ -14,9 +16,11 @@


import org.apache.http.util.EntityUtils;


import org.example.pojo.Currency;


import org.example.server.impl.CurrencySerivceImpl;


import org.example.wsClient.BitgetClient;


import org.example.wsClient.GateClient;


import org.example.wsClient.KucoinClient;


import org.example.wsClient.MexcClient;


import org.json.JSONException;


import org.json.JSONObject;


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.context.annotation.Bean;


@@ -25,8 +29,10 @@


import org.springframework.util.CollectionUtils;




import java.util.ArrayList;


import java.util.HashMap;


import java.util.List;


import java.util.Map;


import java.util.stream.Collectors;




/**


 * @ClassDescription: 客户端请求类


@@ -84,12 +90,9 @@


//    }




    @Bean


    public void kucoinWebsocketRunClientMap() throws Exception {


        List<Currency> mexc = currencyService.list(new LambdaQueryWrapper<Currency>().eq(Currency::getSource, "kucoin"));


    public void bitgetWebsocketRunClientMap() throws JSONException, JsonProcessingException {


        List<Currency> mexc = currencyService.list(new LambdaQueryWrapper<Currency>().eq(Currency::getSource, "bitget"));


        if (!CollectionUtils.isEmpty(mexc)) {


            String result = doPost();


            JSONObject jsonObject = new JSONObject(result);


            String token = jsonObject.getJSONObject("data").getString("token");


            int batchSize = 100; // 每个线程处理的数据量


            int totalSize = mexc.size();


            int threadCount = (int) Math.ceil((double) totalSize / batchSize); // 计算需要的线程数


@@ -98,13 +101,36 @@


                int fromIndex = i * batchSize;


                int toIndex = Math.min(fromIndex + batchSize, totalSize);


                List<Currency> sublist = mexc.subList(fromIndex, toIndex);




                String parameter = getParameter(sublist);


                // 使用自定义线程池提交任务


                threadPoolTaskExecutor.execute(new KucoinClient(sublist,token)::start);


                threadPoolTaskExecutor.execute(new BitgetClient(parameter)::start);


            }




        }


    }


//


//    @Bean


//    public void kucoinWebsocketRunClientMap() throws Exception {


//        List<Currency> mexc = currencyService.list(new LambdaQueryWrapper<Currency>().eq(Currency::getSource, "kucoin"));


//        if (!CollectionUtils.isEmpty(mexc)) {


//            String result = doPost();


//            JSONObject jsonObject = new JSONObject(result);


//            String token = jsonObject.getJSONObject("data").getString("token");


//            int batchSize = 100; // 每个线程处理的数据量


//            int totalSize = mexc.size();


//            int threadCount = (int) Math.ceil((double) totalSize / batchSize); // 计算需要的线程数


//


//            for (int i = 0; i < threadCount; i++) {


//                int fromIndex = i * batchSize;


//                int toIndex = Math.min(fromIndex + batchSize, totalSize);


//                List<Currency> sublist = mexc.subList(fromIndex, toIndex);


//


//                // 使用自定义线程池提交任务


//                threadPoolTaskExecutor.execute(new KucoinClient(sublist,token)::start);


//            }


//


//        }


//    }




    public static String doPost() throws Exception {


        String url = "https://api.kucoin.com/api/v1/bullet-public";


@@ -118,5 +144,28 @@


        defaultHttpClient.getConnectionManager().shutdown();


        return text;


    }




    public String getParameter(List<Currency> list) throws JsonProcessingException, JSONException {


        // 创建一个ObjectMapper实例


        ObjectMapper mapper = new ObjectMapper();


        List<String> symbolList = list.stream().map(Currency::getSymbol).collect(Collectors.toList());


        // 使用Map构建JSON对象


        Map<String, Object> jsonMap = new HashMap<>();


        jsonMap.put("op", "subscribe");


        List<Map<String, String>> mapList = new ArrayList<>();


        symbolList.forEach(f->{


            Map<String, String> argsMap = new HashMap<>();


            argsMap.put("instType", "SPOT");


            argsMap.put("channel", "books15");


            argsMap.put("instId", f);


            mapList.add(argsMap);


        });


        jsonMap.put("args", mapList);




        // 将Map转换为JSON字符串


        String jsonString = mapper.writeValueAsString(jsonMap);


        return jsonString;




    }


}