管理员任务websocket

lxw

2023-05-18 051f52de65df002963aec5edd00478fbd5193bca

管理员任务websocket

 src/main/java/com/whyc/pojo/FaultUpload.java

@@ -43,6 +43,12 @@
    @TableField(exist = false)
    private Integer confirmedNum;

    @TableField(exist = false)
    private Integer pageNum;

    @TableField(exist = false)
    private Integer pageSize;

    public Integer getId() {
        return id;
    }
@@ -210,4 +216,20 @@
    public void setConfirmedNum(Integer confirmedNum) {
        this.confirmedNum = confirmedNum;
    }

    public Integer getPageNum() {
        return pageNum;
    }

    public void setPageNum(Integer pageNum) {
        this.pageNum = pageNum;
    }

    public Integer getPageSize() {
        return pageSize;
    }

    public void setPageSize(Integer pageSize) {
        this.pageSize = pageSize;
    }
}

 src/main/java/com/whyc/webSocket/TaskManageAdminWebsocket.java

New file
@@ -0,0 +1,153 @@
package com.whyc.webSocket;

import com.whyc.config.WebSocketConfig;
import com.whyc.dto.Response;
import com.whyc.factory.ThreadPoolExecutorFactory;
import com.whyc.pojo.FaultUpload;
import com.whyc.pojo.UserInf;
import com.whyc.service.FaultUploadService;
import com.whyc.util.ActionUtil;
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.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
@ServerEndpoint(value = "/taskMAdmin", encoders = WebSocketEncoder.class, configurator = WebSocketConfig.class)
public class TaskManageAdminWebsocket {
    private Session session;

    private Thread thread;

    private volatile boolean runFlag = true;

    private volatile Map<String, Thread> threadMap = new HashMap<>();

    private volatile Map<Long, Boolean> threadFlagMap = new HashMap<>();

    private static final int executeTime = 5000;

    private static FaultUploadService faultUploadService;

    private HttpSession httpSession;

    @Autowired
    public void setFaultUploadService(FaultUploadService faultUploadService) {
        TaskManageAdminWebsocket.faultUploadService = faultUploadService;
    }

    @OnOpen
    public void onOpen(Session session, EndpointConfig config) {
        this.session = session;
        this.httpSession = (HttpSession) config.getUserProperties().get("httpSession");
    }

    @OnMessage
    public void onMessage(Session session, String message) {
        UserInf user = (UserInf) this.httpSession.getAttribute("user");
        final int userId = user.getUId().intValue();
        //final int userId = 1041;
        FaultUpload upload = ActionUtil.getGson().fromJson(message, FaultUpload.class);
        thread = new Thread("Thread_RealTime") {
            @Override
            public void run() {
                while (runFlag && !isInterrupted()) {
                    Thread thread = currentThread();
                    threadFlagMap.put(thread.getId(), true);
                    try {
                        if (session.isOpen()) {
                            //推送信息
                            synchronized (session) {
                                try {
                                    Map<String, Object> res = new HashMap<>();
                                    while (!currentThread().isInterrupted()) {
                                        ThreadPoolExecutor poolExecutor = ThreadPoolExecutorFactory.getPoolExecutor();
                                        CountDownLatch latch = new CountDownLatch(5);
                                        poolExecutor.execute(() -> {
                                            //本月
                                            Response monRes = faultUploadService.groupNameCount(1, userId);
                                            res.put("monRes", monRes);
                                            latch.countDown();
                                        });
                                        poolExecutor.execute(() -> {
                                            //本季度
                                            Response quarterRes = faultUploadService.groupNameCount(2, userId);
                                            res.put("quarterRes", quarterRes);
                                            latch.countDown();
                                        });
                                        poolExecutor.execute(() -> {
                                            //本年
                                            Response yearRes = faultUploadService.groupNameCount(3, userId);
                                            res.put("yearRes", yearRes);
                                            latch.countDown();
                                        });
                                        poolExecutor.execute(() -> {
                                            //隐患故障上报进度-最近一周及最近一月-管理
                                            Response lastPeriodRes = faultUploadService.getListOfLastPeriod(1, userId);
                                            res.put("lastPeriodRes", lastPeriodRes);
                                            latch.countDown();
                                        });
                                        poolExecutor.execute(() -> {
                                            //列表分页-管理
                                            Response listRes = faultUploadService.listPage2(upload.getPageNum(), upload.getPageSize(), upload, userId);
                                            res.put("listRes", listRes);
                                            latch.countDown();
                                        });
                                        latch.await(10, TimeUnit.MINUTES);
                                        session.getBasicRemote().sendObject(new Response().set(1, res));
                                        sleep(4000);
                                    }
                                } catch (Exception e) {
                                    this.interrupt();
                                }
                            }
                            threadFlagMap.put(thread.getId(), false);
                        }
                        sleep(executeTime);
                    } catch (Exception e) {
                        interrupt();
                    }
                }
            }
        };
        thread.start();
        threadFlagMap.put(thread.getId(), true);
        //停止老的socket线程
        Thread threadBefore = threadMap.get(session.getId());
        if (threadBefore != null && threadBefore.isAlive()) {
            while (threadFlagMap.get(threadBefore.getId())) {
            }
            threadBefore.interrupt();
        }
        //将线程存储,便于调用定位
        threadMap.put(session.getId(), this.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());
    }

}