package com.whyc.webSocket; import com.whyc.config.WebSocketConfig; import com.whyc.dto.Response; import com.whyc.factory.ThreadPoolExecutorFactory; import com.whyc.pojo.RtState; import com.whyc.pojo.StationInf; import com.whyc.service.BattInfService; import com.whyc.service.RtDataService; import com.whyc.service.RtStateService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import javax.servlet.http.HttpSession; import javax.websocket.*; import javax.websocket.server.ServerEndpoint; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; /** * packageName com.whyc.websocket * * @author lxw * @version JDK 8 * @className BattSocket (此处以class为例) * @date 2024/6/15 * @description TODO */ @Component @ServerEndpoint(value = "/batt", encoders = WebSocketEncoder.class) public class BattSocket { private Session session; private Thread thread; private volatile boolean runFlag = true; private volatile Map threadMap = new HashMap<>(); private volatile Map threadFlagMap = new HashMap<>(); private static final int executeTime = 5000; private static RtStateService rtStateService; private static RtDataService rtDataService; @Autowired public void setRtStateService(RtStateService rtStateService) { BattSocket.rtStateService = rtStateService; } @Autowired public void setRtDataService(RtDataService rtDataService) { BattSocket.rtDataService = rtDataService; } @OnOpen public void onOpen(Session session) { this.session = session; } @OnMessage public void onMessage(Session session, String message) { int binfId=Integer.valueOf(message); Thread thread = new Thread() { @Override public void run() { try { Map res = new HashMap<>(); while (!currentThread().isInterrupted()) { ThreadPoolExecutor poolExecutor = ThreadPoolExecutorFactory.getPoolExecutor(); CountDownLatch latch = new CountDownLatch(2); poolExecutor.execute(() -> { Response resRtState = rtStateService.getResRtState(binfId); res.put("resRtState", resRtState); latch.countDown(); }); poolExecutor.execute(() -> { Response resRtData = rtDataService.getResRtData(binfId); res.put("resRtData", resRtData); latch.countDown(); }); latch.await(10, TimeUnit.MINUTES); session.getBasicRemote().sendObject(new Response().set(1, res)); sleep(executeTime); } } catch (Exception e) { this.interrupt(); } } }; thread.start(); this.thread = thread; } @OnClose public void onClose(CloseReason closeReason) { System.err.println("closeReason = " + closeReason); runFlag = false; if (thread != null && thread.isAlive()) { thread.interrupt(); } threadMap.remove(session.getId()); } @OnError public void onError(Throwable error) { error.printStackTrace(); if (thread != null && thread.isAlive()) { thread.interrupt(); } threadMap.remove(session.getId()); } }