| | |
| | | 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 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; |
| | |
| | | * 同步gate交易所交易对 |
| | | */ |
| | | @Scheduled(cron = "0 0/40 * * * ?") |
| | | // @Scheduled(cron = "0/10 * * * * ?") |
| | | public void syncCurrency() { |
| | | // 使用Lock来确保同步 |
| | | syncCurrencyLock.lock(); |
| | |
| | | |
| | | getList.parallelStream().forEach(person -> person.setId(StringUtils.remove(person.getId(), "_"))); |
| | | |
| | | // 获取数据库中已有的symbol列表 |
| | | List<Currency> dbList = currencyService.list(); |
| | | Set<String> symbolSet = dbList.stream().map(Currency::getSymbol).collect(Collectors.toSet()); |
| | | currencyService.remove(new LambdaQueryWrapper<Currency>().eq(Currency::getSource,"gate")); |
| | | |
| | | // 比对接口返回的数据和数据库中已有的数据,找出新增的数据 |
| | | List<Currency> saveList = getList.stream() |
| | | .filter(currency -> !symbolSet.contains(currency.getId())) |
| | | .map(currency -> { |
| | | Currency newCurrency = new Currency(); |
| | | newCurrency.setSymbol(currency.getId()); |
| | |
| | | return newCurrency; |
| | | }) |
| | | .collect(Collectors.toList()); |
| | | |
| | | // 批量保存新增数据到数据库 |
| | | if(CollectionUtils.isNotEmpty(saveList)){ |
| | | currencyService.saveBatch(saveList); |