fix(mcwl): 调整
parent
35ec6d7485
commit
2ef2fabc66
|
@ -1,5 +1,6 @@
|
||||||
package com.mcwl.framework.config;
|
package com.mcwl.framework.config;
|
||||||
|
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
import org.springframework.cache.annotation.CachingConfigurerSupport;
|
import org.springframework.cache.annotation.CachingConfigurerSupport;
|
||||||
import org.springframework.cache.annotation.EnableCaching;
|
import org.springframework.cache.annotation.EnableCaching;
|
||||||
import org.springframework.context.annotation.Bean;
|
import org.springframework.context.annotation.Bean;
|
||||||
|
@ -24,6 +25,16 @@ import java.time.Duration;
|
||||||
@EnableCaching
|
@EnableCaching
|
||||||
public class RedisConfig extends CachingConfigurerSupport
|
public class RedisConfig extends CachingConfigurerSupport
|
||||||
{
|
{
|
||||||
|
|
||||||
|
@Value("${spring.redis.host}")
|
||||||
|
private String host;
|
||||||
|
|
||||||
|
@Value("${spring.redis.port}")
|
||||||
|
private int port;
|
||||||
|
|
||||||
|
@Value("${spring.redis.password}")
|
||||||
|
private String password;
|
||||||
|
|
||||||
@Bean
|
@Bean
|
||||||
@SuppressWarnings(value = { "unchecked", "rawtypes" })
|
@SuppressWarnings(value = { "unchecked", "rawtypes" })
|
||||||
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory)
|
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory)
|
||||||
|
@ -80,8 +91,8 @@ public class RedisConfig extends CachingConfigurerSupport
|
||||||
*/
|
*/
|
||||||
@Bean
|
@Bean
|
||||||
public LettuceConnectionFactory redisConnectionFactory() {
|
public LettuceConnectionFactory redisConnectionFactory() {
|
||||||
RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration("1.13.246.108", 6370);
|
RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration(host, port);
|
||||||
standaloneConfig.setPassword(RedisPassword.of("MuYu_Cloud@Redis"));
|
standaloneConfig.setPassword(RedisPassword.of(password));
|
||||||
|
|
||||||
LettuceClientConfiguration lettuceClientConfig = LettuceClientConfiguration.builder()
|
LettuceClientConfiguration lettuceClientConfig = LettuceClientConfiguration.builder()
|
||||||
.commandTimeout(Duration.ofSeconds(10))
|
.commandTimeout(Duration.ofSeconds(10))
|
||||||
|
|
|
@ -30,8 +30,10 @@ public class AuthenticationEntryPointImpl implements AuthenticationEntryPoint, S
|
||||||
throws IOException
|
throws IOException
|
||||||
{
|
{
|
||||||
int code = HttpStatus.UNAUTHORIZED;
|
int code = HttpStatus.UNAUTHORIZED;
|
||||||
if (e.getCause() instanceof RedisCommandTimeoutException) {
|
if (e.getCause() != null) {
|
||||||
ServletUtils.renderString(response, JSON.toJSONString(AjaxResult.warn("网络超时,请检查网络稍后重试")));
|
if (e.getCause() instanceof RedisCommandTimeoutException) {
|
||||||
|
ServletUtils.renderString(response, JSON.toJSONString(AjaxResult.warn("网络超时,请检查网络稍后重试")));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
String msg = StringUtils.format("请求访问:{},认证失败,无法访问系统资源", request.getRequestURI());
|
String msg = StringUtils.format("请求访问:{},认证失败,无法访问系统资源", request.getRequestURI());
|
||||||
ServletUtils.renderString(response, JSON.toJSONString(AjaxResult.error(code, msg)));
|
ServletUtils.renderString(response, JSON.toJSONString(AjaxResult.error(code, msg)));
|
||||||
|
|
|
@ -52,7 +52,6 @@ public class MqttTemplate implements MqttCallbackExtended, DisposableBean {
|
||||||
|
|
||||||
private String clientId;
|
private String clientId;
|
||||||
|
|
||||||
//================= 运行时组件 =================//
|
|
||||||
private MqttAsyncClient client;
|
private MqttAsyncClient client;
|
||||||
private final Map<String, List<IMessageHandler>> topicHandlers = new ConcurrentHashMap<>();
|
private final Map<String, List<IMessageHandler>> topicHandlers = new ConcurrentHashMap<>();
|
||||||
private final ScheduledExecutorService reconnectExecutor = Executors.newSingleThreadScheduledExecutor();
|
private final ScheduledExecutorService reconnectExecutor = Executors.newSingleThreadScheduledExecutor();
|
||||||
|
@ -63,7 +62,7 @@ public class MqttTemplate implements MqttCallbackExtended, DisposableBean {
|
||||||
this.context = context;
|
this.context = context;
|
||||||
}
|
}
|
||||||
|
|
||||||
//================= 生命周期管理 =================//
|
|
||||||
@PostConstruct
|
@PostConstruct
|
||||||
public void init() throws MqttException {
|
public void init() throws MqttException {
|
||||||
if (StringUtils.isBlank(clientId)) {
|
if (StringUtils.isBlank(clientId)) {
|
||||||
|
@ -112,7 +111,10 @@ public class MqttTemplate implements MqttCallbackExtended, DisposableBean {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
//================= 自动重连机制 =================//
|
/**
|
||||||
|
* 重连管理
|
||||||
|
* @param attempt 重连次数
|
||||||
|
*/
|
||||||
private void scheduleReconnect(int attempt) {
|
private void scheduleReconnect(int attempt) {
|
||||||
if (attempt > maxReconnectAttempts) {
|
if (attempt > maxReconnectAttempts) {
|
||||||
log.error("达到最大重连次数: {}", maxReconnectAttempts);
|
log.error("达到最大重连次数: {}", maxReconnectAttempts);
|
||||||
|
@ -132,6 +134,7 @@ public class MqttTemplate implements MqttCallbackExtended, DisposableBean {
|
||||||
client.setCallback(this); // 关键:重新绑定回调
|
client.setCallback(this); // 关键:重新绑定回调
|
||||||
client.reconnect();
|
client.reconnect();
|
||||||
resubscribeAllTopics(); // 重连后立即订阅
|
resubscribeAllTopics(); // 重连后立即订阅
|
||||||
|
log.info("重连成功");
|
||||||
} catch (MqttException e) {
|
} catch (MqttException e) {
|
||||||
log.error("重连失败", e);
|
log.error("重连失败", e);
|
||||||
scheduleReconnect(attempt + 1);
|
scheduleReconnect(attempt + 1);
|
||||||
|
@ -139,7 +142,12 @@ public class MqttTemplate implements MqttCallbackExtended, DisposableBean {
|
||||||
}, delay, TimeUnit.MILLISECONDS);
|
}, delay, TimeUnit.MILLISECONDS);
|
||||||
}
|
}
|
||||||
|
|
||||||
//================= 消息发布 =================//
|
/**
|
||||||
|
* 发布消息
|
||||||
|
* @param topic 主题
|
||||||
|
* @param payload 消息体
|
||||||
|
* @param qos QoS等级
|
||||||
|
*/
|
||||||
@Async
|
@Async
|
||||||
public void publish(String topic, String payload, int qos) {
|
public void publish(String topic, String payload, int qos) {
|
||||||
try {
|
try {
|
||||||
|
@ -164,7 +172,12 @@ public class MqttTemplate implements MqttCallbackExtended, DisposableBean {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
//================= 订阅管理 =================//
|
/**
|
||||||
|
* 订阅主题
|
||||||
|
* @param topicFilter 主题过滤器
|
||||||
|
* @param qos QoS等级
|
||||||
|
* @param handler 消息处理器
|
||||||
|
*/
|
||||||
public void subscribe(String topicFilter, int qos, IMessageHandler handler) {
|
public void subscribe(String topicFilter, int qos, IMessageHandler handler) {
|
||||||
try {
|
try {
|
||||||
if (!client.isConnected()) {
|
if (!client.isConnected()) {
|
||||||
|
@ -198,7 +211,11 @@ public class MqttTemplate implements MqttCallbackExtended, DisposableBean {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
//================= 回调处理 =================//
|
/**
|
||||||
|
* 消息到达处理
|
||||||
|
* @param topic 主题
|
||||||
|
* @param message 消息
|
||||||
|
*/
|
||||||
@Override
|
@Override
|
||||||
public void messageArrived(String topic, MqttMessage message) {
|
public void messageArrived(String topic, MqttMessage message) {
|
||||||
|
|
||||||
|
@ -222,7 +239,7 @@ public class MqttTemplate implements MqttCallbackExtended, DisposableBean {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void deliveryComplete(IMqttDeliveryToken token) {
|
public void deliveryComplete(IMqttDeliveryToken token) {
|
||||||
// QoS处理逻辑(可选)
|
// QoS处理逻辑
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -234,7 +251,9 @@ public class MqttTemplate implements MqttCallbackExtended, DisposableBean {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
//================= 自动注册处理器 =================//
|
/**
|
||||||
|
* 自动注册处理器
|
||||||
|
*/
|
||||||
private void autoRegisterHandlers() {
|
private void autoRegisterHandlers() {
|
||||||
context.getBeansOfType(IMessageHandler.class).values()
|
context.getBeansOfType(IMessageHandler.class).values()
|
||||||
.forEach(handler -> {
|
.forEach(handler -> {
|
||||||
|
|
Loading…
Reference in New Issue