1

zj

2024-08-07 8ed60795a7c278f9699616f72eb05ce49800ba6f

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
package org.example.geteclient.WsBean;
 
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.example.geteclient.GeteClientApplication;
import org.example.geteclient.pojo.Currency;
import org.example.geteclient.server.impl.CurrencySerivceImpl;
import org.example.geteclient.wsClinet.GateClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.SpringApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.CollectionUtils;
 
import java.util.List;
 
/**
 * @ClassDescription: 客户端请求类
 * @JdkVersion: 1.8
 * @Created: 2023/8/31 16:13
 */
@Slf4j
@Configuration
public class GateWsBean {
 
    @Autowired
    private CurrencySerivceImpl currencyService;
 
    @Autowired
    private ConfigurableApplicationContext context;
 
    @Autowired
    @Qualifier("threadPoolTaskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
 
    @Bean
    public void gateWebsocketRunClientMap() {
        List<Currency> mexc = currencyService.list(new LambdaQueryWrapper<Currency>().eq(Currency::getSource, "gate"));
        if (!CollectionUtils.isEmpty(mexc)) {
            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(() -> {
                    try {
                        new GateClient(sublist).start();
                    } catch (Exception e) {
                        run();
                    }
                });
            }
 
        }
    }
 
    private boolean runExecuted = false;
    private synchronized void run() {
 
        if (runExecuted) {
            return; // 已经执行过，直接返回
        }
        runExecuted = true;
        log.info("ws 异常开始重启");
        Thread restartThread = new Thread(() -> {
            try {
                SpringApplication.exit(context, () -> 0);
                SpringApplication.run(GeteClientApplication.class);
                log.info("ws 重启成功");
            } catch (Exception e) {
                e.printStackTrace();
                log.error("ws 重启失败");
            }
        });
        restartThread.setDaemon(false);
        restartThread.start();
        log.info("ws 重启失败");
    }
}