多线程工具类，websocket

ruoyi-common/pom.xml

@@ -184,7 +184,10 @@
             <artifactId>spring-boot-starter-mail</artifactId>
             <version>3.1.4</version>
         </dependency>
-
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
         <dependency>
             <groupId>javax.activation</groupId>
             <artifactId>activation</artifactId>

ruoyi-common/src/main/java/com/ruoyi/common/annotation/VoidFunction.java

@@ -0,0 +1,13 @@
+package com.ruoyi.common.annotation;
+/**
+ * 一个参数、没有返回
+ * @author wangqiong
+ */
+@FunctionalInterface
+public interface VoidFunction<T> {
+    /**
+     * 有一个参数
+     * @param param
+     */
+    void apply(T param);
+}

ruoyi-common/src/main/java/com/ruoyi/common/constant/Constants.java

@@ -129,6 +129,7 @@ public class Constants
      */
     public static final String LOOKUP_LDAPS = "ldaps:";
     public static final String DEFAULT_PASSWORD = "123456";
+    public static final String SPLIT_COMMA =",";
 
     /**
      * 自动识别json对象白名单配置（仅允许解析的包名，范围越小越安全）

ruoyi-common/src/main/java/com/ruoyi/common/utils/thread/MultipleThreadListParam.java

@@ -0,0 +1,22 @@
+package com.ruoyi.common.utils.thread;
+
+import lombok.Data;
+
+import java.util.List;
+import java.util.function.Function;
+
+/**
+ * @author wangqiong
+ * @date 2023/11/25 9:25
+ **/
+@Data
+public class MultipleThreadListParam<T,R> {
+    /**需要执行的方法*/
+    private Function<List<T>,R> function;
+    /**参数ID*/
+    private List<T> list;
+    public MultipleThreadListParam(Function<List<T>,R> function, List<T> list){
+        this.function=function;
+        this.list=list;
+    }
+}

ruoyi-common/src/main/java/com/ruoyi/common/utils/thread/MultipleThreadStringParam.java

@@ -0,0 +1,21 @@
+package com.ruoyi.common.utils.thread;
+
+import lombok.Data;
+
+import java.util.function.Function;
+
+/**
+ * @author wangqiong
+ * @date 2023/11/25 9:25
+ **/
+@Data
+public class MultipleThreadStringParam<T> {
+    /**需要执行的方法*/
+    private Function<String,T> function;
+    /**参数ID*/
+    private String ids;
+    public MultipleThreadStringParam(Function<String,T> function, String ids){
+        this.function=function;
+        this.ids=ids;
+    }
+}

ruoyi-common/src/main/java/com/ruoyi/common/utils/thread/MultipleThreadWorkUtil.java

@@ -0,0 +1,547 @@
+package com.ruoyi.common.utils.thread;
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.util.ArrayUtil;
+import cn.hutool.core.util.StrUtil;
+import com.ruoyi.common.annotation.VoidFunction;
+import com.ruoyi.common.constant.Constants;
+import com.ruoyi.common.exception.ServiceException;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.function.Function;
+
+/**
+ * @description 多线程操作数据库，一个线程异常全部回滚
+ * @Author wangqiong
+ * @Date 2023/11/25 14:02
+ * @Version V1.0
+ **/
+@Slf4j
+public class MultipleThreadWorkUtil {
+    private static  int SIMPLE_TIME_COUNT=1000;
+
+    public static <R,A>List<R> exec(Function<List<A>,R> execFun, List<A> list){
+        List<R> returnList=new ArrayList<>();
+        if(CollectionUtil.isEmpty(list)){
+            return returnList;
+        }
+        if(list.size()<SIMPLE_TIME_COUNT){
+            return Arrays.asList(execFun.apply(list));
+        }
+        int times=list.size()/SIMPLE_TIME_COUNT;
+        if(list.size()%SIMPLE_TIME_COUNT>0){
+            times++;
+        }
+        CountDownLatch mainLatch=new CountDownLatch(1);
+        //监控子线程
+        CountDownLatch threadLatch=new CountDownLatch(times);
+        //根据子线程执行结果判断是否需要回滚
+        BlockingDeque<Boolean>  resultList=new LinkedBlockingDeque<>();
+        //必须使用对象，如果使用变量会造成线程之间不能共享变量值
+        RollBack rollBack=new RollBack(false);
+        ExecutorService executorService=Executors.newFixedThreadPool(times);
+        List<Future<R>> futureList=new ArrayList<>();
+        for (int i = 0; i <times ; i++) {
+            if(i*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT<list.size()){
+                // 创建子线程
+                Future<R> future=executorService.submit(new ExecThread(mainLatch,threadLatch,rollBack,resultList,list.subList(i*SIMPLE_TIME_COUNT,i*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT),execFun));
+                futureList.add(future);
+            }else{
+                Future<R> future=executorService.submit(new ExecThread(mainLatch,threadLatch,rollBack,resultList,list.subList(i*SIMPLE_TIME_COUNT,list.size()),execFun));
+                futureList.add(future);
+            }
+        }
+        setResult(executorService,returnList,futureList,resultList,
+                mainLatch,threadLatch,rollBack,times);
+        return returnList;
+    }
+
+    public static <A>void exec(VoidFunction<Set<A>> execFun, Set<A> set){
+        if(set.size()<SIMPLE_TIME_COUNT){
+            execFun.apply(set);
+            return;
+        }
+        int times=set.size()/SIMPLE_TIME_COUNT;
+        if(set.size()%SIMPLE_TIME_COUNT>0){
+            times++;
+        }
+        CountDownLatch mainLatch=new CountDownLatch(1);
+        //监控子线程
+        CountDownLatch threadLatch=new CountDownLatch(times);
+        //根据子线程执行结果判断是否需要回滚
+        BlockingDeque<Boolean>  resultList=new LinkedBlockingDeque<>();
+        //必须使用对象，如果使用变量会造成线程之间不能共享变量值
+        RollBack rollBack=new RollBack(false);
+        ExecutorService executorService=Executors.newFixedThreadPool(times);
+        List<A> list=new ArrayList<>(set);
+        List<Future> futureList=new ArrayList<>();
+        for (int i = 0; i <times ; i++) {
+            Future future;
+            if(i*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT<list.size()){
+                future=executorService.submit(new VoidExecThread(mainLatch,threadLatch,rollBack,resultList,list.subList(i*SIMPLE_TIME_COUNT,i*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT),execFun));
+            }else{
+                future=executorService.submit(new VoidExecThread(mainLatch,threadLatch,rollBack,resultList,list.subList(i*SIMPLE_TIME_COUNT,list.size()),execFun));
+            }
+            futureList.add(future);
+        }
+        /**存放子线程返回结果*/
+        List<Boolean> backUpResult=new ArrayList<>();
+        try{
+            //
+            boolean await=threadLatch.await(times*3,TimeUnit.SECONDS);
+            if(!await){
+                rollBack.setRollBack(true);
+            }else{
+                //查看执行情况，如果有存在需要回滚的线程，则全部回滚
+                for (int i = 0; i <times ; i++) {
+                    Boolean result=resultList.take();
+                    backUpResult.add(result);
+                    if(result){
+                        /**有线程执行异常，需要回滚子线程*/
+                        rollBack.setRollBack(true);
+                    }
+                }
+            }
+        }catch (InterruptedException e){
+            e.printStackTrace();
+            throw new ServiceException(e.getMessage());
+        }finally {
+            //子线程再次继续执行
+            mainLatch.countDown();
+            executorService.shutdown();
+        }
+        /**检查子线程是否有异常，有异常整体回滚*/
+        for (int i = 0; i <times ; i++) {
+            if(CollectionUtil.isNotEmpty(backUpResult)){
+                Boolean result=backUpResult.get(i);
+                if(result){
+                    /**有线程执行异常，需要回滚子线程*/
+                    throw new ServiceException("多线程执行异常");
+                }
+            }else{
+                throw new ServiceException("多线程执行异常");
+            }
+        }
+        for (Future future : futureList) {
+            try {
+                future.get();
+            } catch (Exception e) {
+                throw new ServiceException(e.getMessage());
+            }
+        }
+    }
+
+   
+
+    public static <R>List<R> execFun(MultipleThreadStringParam<R>...params){
+        List<R> returnList=new ArrayList<>();
+        if(ArrayUtil.isEmpty(params)){
+            return returnList;
+        }
+        List<Integer> threadCountList=new ArrayList<>();
+        for (MultipleThreadStringParam param : params) {
+            List<String> idList= Arrays.asList(param.getIds().split(Constants.SPLIT_COMMA));
+            if(idList.size()<SIMPLE_TIME_COUNT){
+                threadCountList.add(1);
+            }else{
+                if(idList.size()%SIMPLE_TIME_COUNT>0){
+                    threadCountList.add(idList.size()/SIMPLE_TIME_COUNT+1);
+                }else{
+                    threadCountList.add(idList.size()/SIMPLE_TIME_COUNT);
+                }
+            }
+        }
+        int times=0;
+        for (Integer count : threadCountList) {
+            times+=count;
+        }
+        CountDownLatch mainLatch=new CountDownLatch(1);
+        //监控子线程
+        CountDownLatch threadLatch=new CountDownLatch(times);
+        //根据子线程执行结果判断是否需要回滚
+        BlockingDeque<Boolean>  resultList=new LinkedBlockingDeque<>();
+        //必须使用对象，如果使用变量会造成线程之间不能共享变量值
+        RollBack rollBack=new RollBack(false);
+        ExecutorService executorService=Executors.newFixedThreadPool(times);
+        List<Future<R>> futureList=new ArrayList<>();
+        for (int i = 0; i < params.length; i++) {
+            MultipleThreadStringParam param=params[i];
+            List<String> idList= Arrays.asList(param.getIds().split(Constants.SPLIT_COMMA));
+            for (int j = 0; j <threadCountList.get(i) ; j++) {
+                if(j*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT<idList.size()){
+                    Future<R> future=executorService.submit(new ExecByIdsStringThread<>(mainLatch,threadLatch,rollBack,resultList,StrUtil.join(Constants.SPLIT_COMMA,idList.subList(j*SIMPLE_TIME_COUNT,j*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT)),param.getFunction()));
+                    futureList.add(future);
+                }else{
+                    Future<R> future=executorService.submit(new ExecByIdsStringThread(mainLatch,threadLatch,rollBack,resultList,StrUtil.join(Constants.SPLIT_COMMA,idList.subList(j*SIMPLE_TIME_COUNT,idList.size())),param.getFunction()));
+                    futureList.add(future);
+                }
+            }
+        }
+        setResult(executorService,returnList,futureList,resultList,
+                mainLatch,threadLatch,rollBack,times);
+        return returnList;
+    }
+
+    public static <R>List<R> execByIds(Function<String,R> execFun,String ids){
+        List<R> returnList=new ArrayList<>();
+        if(StrUtil.isEmpty(ids)){
+            return returnList;
+        }
+        List<String> idList= Arrays.asList(ids.split(Constants.SPLIT_COMMA));
+        if(idList.size()<SIMPLE_TIME_COUNT){
+            return Arrays.asList(execFun.apply(ids));
+        }
+        int times=idList.size()/SIMPLE_TIME_COUNT;
+        if(idList.size()%SIMPLE_TIME_COUNT>0){
+            times++;
+        }
+        CountDownLatch mainLatch=new CountDownLatch(1);
+        //监控子线程
+        CountDownLatch threadLatch=new CountDownLatch(times);
+        //根据子线程执行结果判断是否需要回滚
+        BlockingDeque<Boolean>  resultList=new LinkedBlockingDeque<>();
+        //必须使用对象，如果使用变量会造成线程之间不能共享变量值
+        RollBack rollBack=new RollBack(false);
+        ExecutorService executorService=Executors.newFixedThreadPool(times);
+        List<Future<R>> futureList=new ArrayList<>();
+        for (int i = 0; i <times ; i++) {
+            if(i*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT<idList.size()){
+                Future<R> future=executorService.submit(new ExecByIdsStringThread<>(mainLatch,threadLatch,rollBack,resultList,StrUtil.join(Constants.SPLIT_COMMA,idList.subList(i*SIMPLE_TIME_COUNT,i*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT)),execFun));
+                futureList.add(future);
+            }else{
+                Future<R> future=executorService.submit(new ExecByIdsStringThread(mainLatch,threadLatch,rollBack,resultList,StrUtil.join(Constants.SPLIT_COMMA,idList.subList(i*SIMPLE_TIME_COUNT,idList.size())),execFun));
+                futureList.add(future);
+            }
+        }
+        setResult(executorService,returnList,futureList,resultList,
+                mainLatch,threadLatch,rollBack,times);
+        return returnList;
+    }
+
+    public static <R>List<R> execByIds(Function<List<String>,List<R>> execFun,List<String> idList){
+        List<R> returnList=new ArrayList<>();
+        if(CollectionUtil.isEmpty(idList)){
+            return returnList;
+        }
+        if(idList.size()<SIMPLE_TIME_COUNT){
+            return execFun.apply(idList);
+        }
+        int times=idList.size()/SIMPLE_TIME_COUNT;
+        if(idList.size()%SIMPLE_TIME_COUNT>0){
+            times++;
+        }
+        CountDownLatch mainLatch=new CountDownLatch(1);
+        //监控子线程
+        CountDownLatch threadLatch=new CountDownLatch(times);
+        //根据子线程执行结果判断是否需要回滚
+        BlockingDeque<Boolean>  resultList=new LinkedBlockingDeque<>();
+        //必须使用对象，如果使用变量会造成线程之间不能共享变量值
+        RollBack rollBack=new RollBack(false);
+        ExecutorService executorService=Executors.newFixedThreadPool(times);
+        List<Future<List<R>>> futureList=new ArrayList<>();
+        for (int i = 0; i <times ; i++) {
+            if(i*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT<idList.size()){
+                Future<List<R>> future=executorService.submit(new ExecByIdsStringListThread<>(mainLatch,threadLatch,rollBack,resultList,idList.subList(i*SIMPLE_TIME_COUNT,i*SIMPLE_TIME_COUNT+SIMPLE_TIME_COUNT),execFun));
+                futureList.add(future);
+            }else{
+                Future<List<R>> future=executorService.submit(new ExecByIdsStringListThread(mainLatch,threadLatch,rollBack,resultList,idList.subList(i*SIMPLE_TIME_COUNT,idList.size()),execFun));
+                futureList.add(future);
+            }
+        }
+        /**存放子线程返回结果*/
+        List<Boolean> backUpResult=new ArrayList<>();
+        try{
+            //
+            boolean await=threadLatch.await(times*3,TimeUnit.SECONDS);
+            if(!await){
+                rollBack.setRollBack(true);
+            }else{
+                //查看执行情况，如果有存在需要回滚的线程，则全部回滚
+                for (int i = 0; i <times ; i++) {
+                    Boolean result=resultList.take();
+                    backUpResult.add(result);
+                    if(result){
+                        /**有线程执行异常，需要回滚子线程*/
+                        rollBack.setRollBack(true);
+                    }
+                }
+            }
+        }catch (InterruptedException e){
+            e.printStackTrace();
+            throw new ServiceException("多线程执行异常");
+        }finally {
+            //子线程再次继续执行
+            mainLatch.countDown();
+            executorService.shutdown();
+        }
+        /**检查子线程是否有异常，有异常整体回滚*/
+        for (int i = 0; i <times ; i++) {
+            if(CollectionUtil.isNotEmpty(backUpResult)){
+                Boolean result=backUpResult.get(i);
+                if(result){
+                    /**有线程执行异常，需要回滚子线程*/
+                    throw new ServiceException("多线程执行异常");
+                }
+            }else{
+                throw new ServiceException("多线程执行异常");
+            }
+        }
+        for (Future<List<R>> future : futureList) {
+            try {
+                returnList.addAll(future.get());
+            } catch (Exception e) {
+                throw new ServiceException("多线程执行异常");
+            }
+        }
+        return returnList;
+    }
+
+    private static <R>void setResult(ExecutorService executorService,List<R> returnList,List<Future<R>> futureList,BlockingDeque<Boolean>  resultList
+            ,CountDownLatch mainLatch,CountDownLatch threadLatch,RollBack rollBack,int times){
+        /**存放子线程返回结果*/
+        List<Boolean> backUpResult=new ArrayList<>();
+        try{
+            //
+            boolean await=threadLatch.await(times*3,TimeUnit.SECONDS);
+            if(!await){
+                rollBack.setRollBack(true);
+            }else{
+                //查看执行情况，如果有存在需要回滚的线程，则全部回滚
+                for (int i = 0; i <times ; i++) {
+                    Boolean result=resultList.take();
+                    backUpResult.add(result);
+                    if(result){
+                        /**有线程执行异常，需要回滚子线程*/
+                        rollBack.setRollBack(true);
+                    }
+                }
+            }
+        }catch (InterruptedException e){
+            e.printStackTrace();
+            throw new ServiceException("多线程执行异常");
+        }finally {
+            //子线程再次继续执行
+            mainLatch.countDown();
+            executorService.shutdown();
+        }
+        /**检查子线程是否有异常，有异常整体回滚*/
+        for (int i = 0; i <times ; i++) {
+            if(CollectionUtil.isNotEmpty(backUpResult)){
+                Boolean result=backUpResult.get(i);
+                if(result){
+                    /**有线程执行异常，需要回滚子线程*/
+                    throw new ServiceException("多线程执行异常");
+                }
+            }else{
+                throw new ServiceException("多线程执行异常");
+            }
+        }
+        for (Future<R> future : futureList) {
+            try {
+                returnList.add(future.get());
+            } catch (Exception e) {
+                throw new ServiceException("多线程执行异常");
+            }
+        }
+    }
+
+    static class QueryThread<R> implements Callable<List<R>>{
+        private String ids;
+        private Function<String,List<R>> execFun;
+        public QueryThread(String ids,Function<String,List<R>> execFun){
+            this.ids=ids;
+            this.execFun=execFun;
+        }
+        @Override
+        public List<R> call(){
+            return execFun.apply(ids);
+        }
+    }
+
+    static class ExecThread<T,R> implements Callable<R>{
+        /**主线程监控*/
+        private CountDownLatch mainLatch;
+        /**子线程监控*/
+        private CountDownLatch threadLatch;
+        /**是否回滚*/
+        private RollBack rollBack;
+        private BlockingDeque<Boolean>  resultList;
+        private List<T> list;
+        private Function<List<T>,R> execFun;
+        public ExecThread(CountDownLatch mainLatch,CountDownLatch threadLatch,RollBack rollBack,BlockingDeque<Boolean>  resultList,List<T> list,Function<List<T>,R> execFun){
+            this.mainLatch=mainLatch;
+            this.threadLatch=threadLatch;
+            this.rollBack=rollBack;
+            this.resultList=resultList;
+            this.list=list;
+            this.execFun=execFun;
+        }
+        @Override
+        @Transactional(rollbackFor = Exception.class)
+        public R call(){
+            // 是否回滚
+            Boolean result=false;
+            R r=null;
+            try{
+                // 对数据库进行操作
+                r=execFun.apply(list);
+            }catch (Exception e){
+                e.printStackTrace();
+                result=true;
+            }
+            resultList.add(result);
+            // 子线程-1，切换到主线程执行
+            threadLatch.countDown();
+            try{
+                // 等待主线程执行
+                mainLatch.await();
+            }catch (InterruptedException e){
+                throw new ServiceException("多线程执行异常");
+            }
+            if(rollBack.getRollBack()){
+                throw new ServiceException("多线程执行异常");
+            }
+            return r;
+        }
+    }
+
+    static class VoidExecThread<T> implements Runnable{
+        /**主线程监控*/
+        private CountDownLatch mainLatch;
+        /**子线程监控*/
+        private CountDownLatch threadLatch;
+        /**是否回滚*/
+        private RollBack rollBack;
+        private BlockingDeque<Boolean>  resultList;
+        private List<T> list;
+        private VoidFunction<Set<T>> execFun;
+        public VoidExecThread(CountDownLatch mainLatch,CountDownLatch threadLatch,RollBack rollBack,BlockingDeque<Boolean>  resultList,List<T> list,VoidFunction<Set<T>> execFun){
+            this.mainLatch=mainLatch;
+            this.threadLatch=threadLatch;
+            this.rollBack=rollBack;
+            this.resultList=resultList;
+            this.list=list;
+            this.execFun=execFun;
+        }
+        @Override
+        public void run() {
+            Boolean result=false;
+            try{
+                execFun.apply(new HashSet<>(list));
+            }catch (Exception e){
+                e.printStackTrace();
+                result=true;
+            }
+            resultList.add(result);
+            threadLatch.countDown();
+            try{
+                mainLatch.await();
+            }catch (InterruptedException e){
+                throw new ServiceException("多线程执行异常");
+            }
+            if(rollBack.getRollBack()){
+                throw new ServiceException("多线程执行异常");
+            }
+        }
+    }
+
+
+    static class ExecByIdsStringThread<R> implements Callable<R>{
+        /**主线程监控*/
+        private CountDownLatch mainLatch;
+        /**子线程监控*/
+        private CountDownLatch threadLatch;
+        /**是否回滚*/
+        private RollBack rollBack;
+        private BlockingDeque<Boolean>  resultList;
+        private String ids;
+        private Function<String,R> execFun;
+        public ExecByIdsStringThread(CountDownLatch mainLatch,CountDownLatch threadLatch,RollBack rollBack,BlockingDeque<Boolean>  resultList,String ids,Function<String,R> execFun){
+            this.mainLatch=mainLatch;
+            this.threadLatch=threadLatch;
+            this.rollBack=rollBack;
+            this.resultList=resultList;
+            this.ids=ids;
+            this.execFun=execFun;
+        }
+        @Override
+        public R call(){
+            Boolean result=false;
+            R r=null;
+            try{
+                r=execFun.apply(ids);
+            }catch (Exception e){
+                e.printStackTrace();
+                result=true;
+            }
+            resultList.add(result);
+            threadLatch.countDown();
+            try{
+                mainLatch.await();
+            }catch (InterruptedException e){
+                throw new ServiceException("多线程执行异常");
+            }
+            if(rollBack.getRollBack()){
+                throw new ServiceException("多线程执行异常");
+            }
+            return r;
+        }
+    }
+
+
+
+
+    static class ExecByIdsStringListThread<R> implements Callable<R>{
+        /**主线程监控*/
+        private CountDownLatch mainLatch;
+        /**子线程监控*/
+        private CountDownLatch threadLatch;
+        /**是否回滚*/
+        private RollBack rollBack;
+        private BlockingDeque<Boolean>  resultList;
+        private List<String> ids;
+        private Function<List<String>,R> execFun;
+        public ExecByIdsStringListThread(CountDownLatch mainLatch,CountDownLatch threadLatch,RollBack rollBack,BlockingDeque<Boolean>  resultList,List<String> ids,Function<List<String>,R> execFun){
+            this.mainLatch=mainLatch;
+            this.threadLatch=threadLatch;
+            this.rollBack=rollBack;
+            this.resultList=resultList;
+            this.ids=ids;
+            this.execFun=execFun;
+        }
+        @Override
+        public R call(){
+            Boolean result=false;
+            R r=null;
+            try{
+                r=execFun.apply(ids);
+            }catch (Exception e){
+                e.printStackTrace();
+                result=true;
+            }
+            resultList.add(result);
+            threadLatch.countDown();
+            try{
+                mainLatch.await();
+            }catch (InterruptedException e){
+                throw new ServiceException("多线程执行异常");
+            }
+            if(rollBack.getRollBack()){
+                throw new ServiceException("多线程执行异常");
+            }
+            return r;
+        }
+    }
+
+    @Data
+    static class RollBack{
+        private Boolean rollBack;
+        public RollBack(Boolean rollBack){
+            this.rollBack=rollBack;
+        }
+    }
+}

ruoyi-common/src/main/java/com/ruoyi/common/websocket/MyWebSocketHandler.java

@@ -0,0 +1,140 @@
+package com.ruoyi.common.websocket;
+
+
+import com.ruoyi.common.exception.ServiceException;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.websocket.*;
+import javax.websocket.server.PathParam;
+import javax.websocket.server.ServerEndpoint;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+/**
+ * @author wangqiong
+ * @description
+ * @date 2023-11-25 15:16
+ */
+@Component
+@Slf4j
+@ServerEndpoint("/websocket/{userName}")
+public class MyWebSocketHandler {
+    // 接口路径 ws://127.0.0.1:9000/websocket;
+    private Session session;
+
+    //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。
+    private static CopyOnWriteArraySet<Session> sessions  = new CopyOnWriteArraySet<>();
+    // 用来存在线连接数
+    private static Map<String, Session> sessionPool = new HashMap<>();
+
+    /**
+     * 链接成功调用的方法
+     */
+    @OnOpen
+    public void onOpen(Session session, @PathParam(value = "userName") String userName) {
+        try {
+            sessions.add(session);
+            sessionPool.put(userName, session);
+
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 链接关闭调用的方法
+     */
+    @OnClose
+    public void onClose(Session session,@PathParam(value = "userName") String userName) {
+        try {
+            if(session!=null && session.isOpen()){
+                session.close();
+                sessions.remove(session);
+                sessionPool.remove(userName);
+            }
+
+            log.info("【websocket消息】连接断开，总数为:" + sessions.size());
+
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 收到客户端消息后调用的方法
+     *
+     * @param message
+     * @param
+     */
+    @OnMessage
+    public void onMessage(@PathParam(value = "userName") String userName, String message) {
+
+        System.out.println("【websocket消息】收到客户端消息:" + message);
+        // 将消息广播给其它用户
+        for (Session session : sessions) {
+            if(session!=null && session.isOpen()){
+                try {
+                    session.getBasicRemote().sendText(message);
+                }catch (Exception e){
+                    throw new ServiceException(e.getMessage());
+                }
+            }else {
+                try {
+                   session.close();
+                } catch (IOException e) {
+                    throw new ServiceException(e.getMessage());
+                }
+               sessions.remove(session);
+                sessionPool.remove(userName);
+            }
+        }
+
+    }
+
+    /**
+     * 发送错误时的处理
+     *
+     * @param session
+     * @param error
+     */
+    @OnError
+    public void onError(Session session, Throwable error) {
+
+        throw new ServiceException(error.getMessage());
+    }
+
+
+    /**
+     * 推消息给前端
+     *
+     * @param userId
+     * @param message
+     * @return
+     */
+    public static Runnable sendOneMessage(String userId, String message) {
+        Session session = sessionPool.get(userId);
+        if (session != null && session.isOpen()) {
+            try {
+                log.info("【推给前端消息】 :" + message);
+
+                //高并发下，防止session占用期间，被其他线程调用
+                synchronized (session) {
+                    session.getBasicRemote().sendText(message);
+                }
+
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+        return null;
+    }
+
+
+
+
+
+}
+

ruoyi-common/src/main/java/com/ruoyi/common/websocket/WebSocketConfig.java

@@ -0,0 +1,26 @@
+package com.ruoyi.common.websocket;//package com.ruoyi.common.websocket;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.server.standard.ServerEndpointExporter;
+
+/**
+ * @author wangqiong
+ * @description
+ * @date 2023-11-25 15:19
+ */
+
+@Configuration
+public class WebSocketConfig {
+
+    /**
+     * 注入ServerEndpointExporter，
+     * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
+     */
+
+    @Bean
+    public ServerEndpointExporter serverEndpointExporter() {
+        return new ServerEndpointExporter();
+    }
+
+}

ruoyi-framework/src/main/java/com/ruoyi/framework/config/SecurityConfig.java

@@ -111,7 +111,7 @@ public class SecurityConfig extends WebSecurityConfigurerAdapter
                 // 过滤请求
                 .authorizeRequests()
                 // 对于登录login 注册register 验证码captchaImage 允许匿名访问
-                .antMatchers("/login", "/register", "/captchaImage","/uploadPath/**").permitAll()
+                .antMatchers("/login", "/register", "/captchaImage","/uploadPath/**","/websocket/**").permitAll()
                 // 静态资源，可匿名访问
                 .antMatchers(HttpMethod.GET, "/", "/*.html", "/**/*.html", "/**/*.css", "/**/*.js", "/profile/**").permitAll()
                 .antMatchers("/swagger-ui.html", "/swagger-resources/**", "/webjars/**", "/*/api-docs", "/druid/**").permitAll()