SH-Arbitrate/Arbitrate-Backend
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'wq' of SH-Arbitrate/Arbitrate-Backend into dev

Browse Source
This commit was merged in pull request #242.
This commit is contained in:
wangqiong123
2023-11-25 18:15:59 +08:00
committed by Gitea
parent aef4ae747e dbefa0b24e
commit f4e47e3348
9 changed files with 775 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ruoyi-common/pom.xml
+4 -1
View File
@@ -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
+13
View File
@@ -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
+1
View File
@@ -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
+22
View File
@@ -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
+21
View File
@@ -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
+547
View File
@@ -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
+140
View File
@@ -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
+26
View File
@@ -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
+1 -1
View File
@@ -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()