修改：会话消息独立线程池；线程池参数统一设置

liusuyi

2024-08-01 60211f59d2c85053533ed151adb2bdc5348dd342

修改：会话消息独立线程池；线程池参数统一设置

 ard-work/src/main/java/com/ruoyi/alarm/config/AsyncConfiguration.java

@@ -76,11 +76,11 @@
    public Executor globalExecutor(){
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(10);
        executor.setCorePoolSize(corePoolSize);
        //配置最大线程数
        executor.setMaxPoolSize(15);
        executor.setMaxPoolSize(maxPoolSize);
        //配置队列大小
        executor.setQueueCapacity(25);
        executor.setQueueCapacity(queueCapacity);
        //线程的名称前缀
        executor.setThreadNamePrefix("globalExecutor-");
        //线程活跃时间（秒）
@@ -93,5 +93,25 @@
        executor.initialize();
        return executor;
    }

    @Bean("msgExecutor")
    public Executor msgExecutor(){
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(corePoolSize);
        //配置最大线程数
        executor.setMaxPoolSize(maxPoolSize);
        //配置队列大小
        executor.setQueueCapacity(queueCapacity);
        //线程的名称前缀
        executor.setThreadNamePrefix("msgExecutor-");
        //线程活跃时间（秒）
        executor.setKeepAliveSeconds(keepAliveSeconds);
        //等待所有任务结束后再关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        //设置拒绝策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //执行初始化
        executor.initialize();
        return executor;
    }
}

 ard-work/src/main/java/com/ruoyi/call/listener/MsgListener.java

@@ -26,7 +26,7 @@
@Slf4j(topic = "msgListener")
public class MsgListener {

    @Async("globalExecutor")
    @Async("msgExecutor")
    @EventListener(MessageEvent.class)
    public void ArdCallHistoryEventListener(MessageEvent messageEvent) {
        log.debug("监听到会话消息事件:"+messageEvent.getArdCallHistory().getContent());

 ard-work/src/main/java/com/ruoyi/call/service/impl/ArdCallHistoryServiceImpl.java

@@ -16,6 +16,7 @@
import com.ruoyi.common.utils.uuid.IdUtils;
import com.ruoyi.system.service.ISysUserService;
import com.ruoyi.utils.websocket.util.WebSocketUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
@@ -42,6 +43,7 @@
 * @date 2024-07-03
 */
@Service
@Slf4j(topic = "msgListener")
public class ArdCallHistoryServiceImpl implements IArdCallHistoryService {
    @Resource
    private ArdCallHistoryMapper ardCallHistoryMapper;

 ard-work/src/main/java/com/ruoyi/utils/mqtt/MqttConsumerCallback.java

@@ -1,5 +1,6 @@
package com.ruoyi.utils.mqtt;

import com.ruoyi.alarm.global.service.IGlobalAlarmService;
import com.ruoyi.alarm.global.service.impl.GlobalAlarmServiceImpl;
import com.ruoyi.alarm.radar.service.ArdRadarService;
import com.ruoyi.common.utils.spring.SpringUtils;
@@ -73,8 +74,9 @@
            // subscribe后得到的消息会执行到这里面
            log.debug("接收消息 【主题】:" + topic + " 【内容】:" + new String(message.getPayload(), StandardCharsets.UTF_8));
            //进行业务处理(接收报警数据)
            GlobalAlarmServiceImpl globalAlarmService = SpringUtils.getBean(GlobalAlarmServiceImpl.class);
            IGlobalAlarmService globalAlarmService = SpringUtils.getBean(IGlobalAlarmService.class);
            globalAlarmService.receiveAlarm(topic, new String(message.getPayload(), StandardCharsets.UTF_8));

            if (topic.equals("minioEvent"))
            {
                IStorageMinioEventService storageMinioEventService = SpringUtils.getBean(IStorageMinioEventService.class);

 ard-work/src/main/java/com/ruoyi/utils/websocket/util/WebSocketUtils.java

@@ -23,7 +23,7 @@
public final class WebSocketUtils {

    // 存储 websocket session
    public static final Map<String, Session> ONLINE_USER_SESSIONS = new ConcurrentHashMap<>();
    public static final ConcurrentMap<String, Session> ONLINE_USER_SESSIONS = new ConcurrentHashMap<>();
    //存储房间
    public static final ConcurrentHashMap<String, Set<String>> ROOM_USER_SET = new ConcurrentHashMap<>();