init

ruoyi-admin/src/test/java/com/ruoyi/test/CaiUnitTest.java

@@ -0,0 +1,13 @@
+package com.ruoyi.test;
+
+import org.springframework.boot.test.context.SpringBootTest;
+
+/**
+ * 带参数单元测试案例
+ *
+ * @author Lion Li
+ */
+@SpringBootTest
+public class CaiUnitTest {
+
+}

ruoyi-cai/src/main/java/com/ruoyi/cai/controller/app/ImAppController.java

@@ -17,7 +17,7 @@ import org.springframework.web.bind.annotation.RestController;
 @RestController
 @RequestMapping("/api/im")
 @Tag(name = "IM相关的接口")
-public class ImController {
+public class ImAppController {
 
     @Autowired
     private ImService imService;

ruoyi-cai/src/main/java/com/ruoyi/cai/controller/app/UserMemberAppController.java

@@ -9,5 +9,5 @@ import org.springframework.web.bind.annotation.RestController;
 @RequestMapping("/api/user/member")
 @Slf4j
 @Tag(name = "用户会员接口")
-public class UserMemberController {
+public class UserMemberAppController {
 }

ruoyi-cai/src/main/java/com/ruoyi/cai/ws/handler/message/CancelMessageHandler.java

@@ -3,6 +3,7 @@ package com.ruoyi.cai.ws.handler.message;
 import com.alibaba.fastjson2.JSONObject;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.ruoyi.cai.domain.UserCall;
+import com.ruoyi.cai.executor.ExecutorConstant;
 import com.ruoyi.cai.service.UserCallService;
 import com.ruoyi.cai.trd.ImDataRes;
 import com.ruoyi.cai.trd.ImMsgGen;
@@ -51,7 +52,7 @@ public class CancelMessageHandler extends AbstractMessageHandle implements IMess
         Long receiverId = room.getReceiverUserData().getId();
         Long callerId = room.getCallUserData().getId();
         ImDataRes imDataRes = ImMsgGen.callNotice(1, callerId, receiverId, 0);
-        yunxin.sendTo(receiverId,callerId,imDataRes);
+        yunxin.sendToSync(receiverId,callerId,imDataRes);
         // 更新房间状态
         userCallService.update(Wrappers.lambdaUpdate(UserCall.class)
                 .eq(UserCall::getId,roomId)

ruoyi-cai/src/main/java/com/ruoyi/cai/ws/handler/message/RefuseMessageHandler.java

@@ -47,7 +47,7 @@ public class RefuseMessageHandler extends AbstractMessageHandle implements IMess
         Long receiverId = room.getReceiverUserData().getId();
         Long callerId = room.getCallUserData().getId();
         ImDataRes imDataRes = ImMsgGen.callNotice(1, callerId, receiverId, 0);
-        yunxin.sendTo(receiverId,callerId,imDataRes);
+        yunxin.sendToSync(receiverId,callerId,imDataRes);
 
         // 更新房间状态
         userCallService.update(Wrappers.lambdaUpdate(UserCall.class)

ruoyi-cai/src/main/java/com/ruoyi/cai/ws/job/CheckTimeOutJob.java

@@ -68,7 +68,7 @@ public class CheckTimeOutJob {
                         .set(UserCall::getStatus, RoomStatusEnums.STATUS_TIMEOUT_CANCEL.getCode()));
                 roomService.closeAllFd(roomId);
                 ImDataRes imDataRes = ImMsgGen.callNotice(3, callUserData.getId(), receiverUserData.getId(), 0);
-                yunxin.sendTo(receiverUserData.getId(),callUserData.getId(),imDataRes);
+                yunxin.sendToSync(receiverUserData.getId(),callUserData.getId(),imDataRes);
             }
         }
     }

ruoyi-cai/src/main/java/com/ruoyi/cai/ws/service/SettleService.java

@@ -156,13 +156,13 @@ public class SettleService {
         Long receiveId = room.getReceiverUserData().getId();
         if(callTime > 0){
             ImDataRes imDataRes = ImMsgGen.callNotice(4, callId, receiveId, callTime);
-            yunxin.sendTo(callId,receiveId,imDataRes);
+            yunxin.sendToSync(callId,receiveId,imDataRes);
         }
         // 接收方通知
         Long anchorIncome = resp.getAnchorIncome();
         if(anchorIncome > 0){
             ImDataRes imDataRes = ImMsgGen.callNotice(4, receiveId, callId, callTime);
-            yunxin.sendTo(receiveId,callId,imDataRes);
+            yunxin.sendToSync(receiveId,callId,imDataRes);
             // 收入通知
 //            ImMsgGen.videoIncome();
 //            yunxin.sendTo(receiveId,null,)

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/Yunxin.java

@@ -2,6 +2,7 @@ package com.ruoyi.yunxin;
 
 import com.alibaba.fastjson2.JSON;
 import com.ruoyi.yunxin.client.ImMessageClient;
+import com.ruoyi.yunxin.config.YunxinExecutorConstant;
 import com.ruoyi.yunxin.config.YunxinProperties;
 import com.ruoyi.yunxin.req.Option;
 import com.ruoyi.yunxin.req.SendAttachMsgReq;
@@ -26,6 +27,12 @@ public class Yunxin {
     @Resource
     private ImMessageClient messageClient;
 
+    public void sendToSync(Long toUid,Long fromUid,Object data){
+        YunxinExecutorConstant.COMMON_EXECUTOR.execute(() -> {
+            this.sendTo(toUid,fromUid,data);
+        });
+    }
+
     public YxDataR<SendMsgResp> sendTo(Long toUid,Long fromUid,Object data){
         SendMsgReq req = new SendMsgReq();
         req.setFrom(fromUid == null ? yunxinProperties.getDefaultFromUid() : fromUid+"");

ruoyi-yunxin/src/main/java/com/ruoyi/yunxin/config/YunxinExecutorConstant.java

@@ -0,0 +1,38 @@
+package com.ruoyi.yunxin.config;
+
+import com.alibaba.ttl.threadpool.TtlExecutors;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import java.util.concurrent.*;
+
+/**
+ * 线程变量定义
+ * <p>created on 2023/3/3 11:16</p>
+ * @author ZL
+ */
+public class YunxinExecutorConstant {
+
+    private final static int CPU_NUM = Runtime.getRuntime().availableProcessors();
+    public static Executor COMMON_EXECUTOR;
+
+    static {
+        ThreadPoolExecutor commonExecutor = new ThreadPoolExecutor(CPU_NUM,
+                CPU_NUM << 2,
+                5,
+                TimeUnit.SECONDS,
+                new ArrayBlockingQueue<>(50),
+                init("commonThreadPool-%d"),
+                new ThreadPoolExecutor.CallerRunsPolicy());
+        COMMON_EXECUTOR = TtlExecutors.getTtlExecutor(commonExecutor);
+    }
+
+    private static ThreadFactory init(String nameFormat){
+        return new ThreadFactoryBuilder().setNameFormat(nameFormat).build();
+    }
+
+    private static ThreadPoolExecutor initExecutor(int corePoolSize, int maxPoolSize, int keepAliveTime, TimeUnit timeUnit,
+                                         BlockingQueue<Runnable> workQueue, String nameFormat){
+        return new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, timeUnit, workQueue,
+                init(nameFormat));
+    }
+}