package org.example.task; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.CollectionUtils; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.reflect.TypeToken; import com.google.gson.Gson; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang.StringUtils; import org.example.pojo.Currency; import org.example.pojo.bo.CurrencyGateBo; import org.example.pojo.bo.CurrencyMexcBo; import org.example.server.impl.CurrencySerivceImpl; import org.example.common.MarketDataClient; import org.example.util.ConverterUtil; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import java.io.BufferedReader; import java.io.InputStreamReader; import java.net.HttpURLConnection; import java.net.URL; import java.util.*; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; /** * @program: demo * @description: gate交易所 * @create: 2024-07-15 17:23 **/ @Component @Slf4j public class GateStock { @Autowired private CurrencySerivceImpl currencyService; private final Lock syncCurrencyLock = new ReentrantLock(); /** * 同步gate交易所交易对 */ @Scheduled(cron = "0 0/40 * * * ?") // @Scheduled(cron = "0/10 * * * * ?") public void syncCurrency() { // 使用Lock来确保同步 syncCurrencyLock.lock(); log.info("【同步gate交易所交易对】---->开始"); try { sync(); } catch (Exception e) { log.error("【同步gate交易所交易对】出现异常: " + e.getMessage()); e.printStackTrace(); } finally { syncCurrencyLock.unlock(); log.info("【同步gate交易所交易对】---->结束"); } } public void sync() throws JsonProcessingException { // 调用外部接口获取数据 String json = MarketDataClient.doGet("https://api.gateio.ws/api/v4/spot/currency_pairs"); // 对返回数据格式进行校验 if (json != null && !json.isEmpty()) { Gson gson = new Gson(); List getList = gson.fromJson(json, new TypeToken>() { }.getType()); getList.parallelStream().forEach(person -> person.setId(StringUtils.remove(person.getId(), "_"))); currencyService.remove(new LambdaQueryWrapper().eq(Currency::getSource,"gate")); // 比对接口返回的数据和数据库中已有的数据,找出新增的数据 List saveList = getList.stream() .map(currency -> { Currency newCurrency = new Currency(); newCurrency.setSymbol(currency.getId()); newCurrency.setBaseAsset(currency.getBase()); newCurrency.setQuoteAsset(currency.getQuote()); newCurrency.setSource(currency.getSource()); return newCurrency; }) .collect(Collectors.toList()); // 批量保存新增数据到数据库 if(CollectionUtils.isNotEmpty(saveList)){ currencyService.saveBatch(saveList); } } else { log.info("外部接口返回数据为空"); } } }