|
|
|
|
@ -25,69 +25,65 @@ package cc.iotkit.ruleengine.handler;
|
|
|
|
|
|
|
|
|
|
import cc.iotkit.common.constant.Constants;
|
|
|
|
|
import cc.iotkit.common.thing.ThingModelMessage;
|
|
|
|
|
import cc.iotkit.common.utils.JsonUtils;
|
|
|
|
|
import cc.iotkit.common.utils.ThreadUtil;
|
|
|
|
|
import cc.iotkit.mq.ConsumerHandler;
|
|
|
|
|
import cc.iotkit.mq.MqConsumer;
|
|
|
|
|
import com.google.common.util.concurrent.ThreadFactoryBuilder;
|
|
|
|
|
import lombok.SneakyThrows;
|
|
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
|
|
import org.springframework.beans.BeansException;
|
|
|
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
|
|
|
import org.springframework.context.ApplicationContext;
|
|
|
|
|
import org.springframework.context.ApplicationContextAware;
|
|
|
|
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
|
|
|
|
|
|
|
|
|
import javax.annotation.PreDestroy;
|
|
|
|
|
import java.util.ArrayList;
|
|
|
|
|
import java.util.HashMap;
|
|
|
|
|
import java.util.List;
|
|
|
|
|
import java.util.Map;
|
|
|
|
|
import java.util.concurrent.LinkedBlockingQueue;
|
|
|
|
|
import java.util.concurrent.ScheduledThreadPoolExecutor;
|
|
|
|
|
import java.util.concurrent.ThreadPoolExecutor;
|
|
|
|
|
import java.util.concurrent.TimeUnit;
|
|
|
|
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
|
|
|
|
|
|
|
|
@Slf4j
|
|
|
|
|
public class RuleDeviceConsumer implements ConsumerHandler<ThingModelMessage>, ApplicationContextAware {
|
|
|
|
|
|
|
|
|
|
private final List<DeviceMessageHandler> handlers = new ArrayList<>();
|
|
|
|
|
private ScheduledThreadPoolExecutor messageHandlerPool;
|
|
|
|
|
private ThreadPoolExecutor messageHandlerPool;
|
|
|
|
|
|
|
|
|
|
/** 连续失败计数器 */
|
|
|
|
|
private final AtomicInteger consecutiveFailures = new AtomicInteger(0);
|
|
|
|
|
/** 熔断打开截止时间戳，0 表示关闭 */
|
|
|
|
|
private volatile long circuitOpenUntil = 0;
|
|
|
|
|
|
|
|
|
|
private static final int FAILURE_THRESHOLD = 10;
|
|
|
|
|
private static final long CIRCUIT_COOLDOWN_MS = 30_000;
|
|
|
|
|
private static final int QUEUE_CAPACITY = 500;
|
|
|
|
|
|
|
|
|
|
//private ThreadPoolExecutor messageHandlerPool;
|
|
|
|
|
@SneakyThrows
|
|
|
|
|
public RuleDeviceConsumer(MqConsumer<ThingModelMessage> consumer) {
|
|
|
|
|
consumer.consume(Constants.THING_MODEL_MESSAGE_TOPIC, this);
|
|
|
|
|
}
|
|
|
|
|
/* @Override
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
|
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
|
|
|
|
|
Map<String, DeviceMessageHandler> handlerMap = applicationContext.getBeansOfType(DeviceMessageHandler.class);
|
|
|
|
|
|
|
|
|
|
// 动态计算核心线程数
|
|
|
|
|
int corePoolSize = Math.max(4, Runtime.getRuntime().availableProcessors());
|
|
|
|
|
// corePoolSize = corePoolSize > handlerMap.size()?corePoolSize:handlerMap.size();
|
|
|
|
|
int maxPoolSize = corePoolSize * 2;
|
|
|
|
|
|
|
|
|
|
int coreSize = handlerMap.size();
|
|
|
|
|
messageHandlerPool = new ThreadPoolExecutor(
|
|
|
|
|
corePoolSize,
|
|
|
|
|
maxPoolSize,
|
|
|
|
|
coreSize,
|
|
|
|
|
coreSize * 2,
|
|
|
|
|
60L, TimeUnit.SECONDS,
|
|
|
|
|
new LinkedBlockingQueue<>(10), // 限制队列大小
|
|
|
|
|
new ThreadFactoryBuilder().setNameFormat("msg-handler-%d").build(),
|
|
|
|
|
new ThreadPoolExecutor.CallerRunsPolicy() // 饱和策略
|
|
|
|
|
new LinkedBlockingQueue<>(QUEUE_CAPACITY),
|
|
|
|
|
new ThreadPoolExecutor.CallerRunsPolicy()
|
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
this.handlers.addAll(handlerMap.values());
|
|
|
|
|
}*/
|
|
|
|
|
@Override
|
|
|
|
|
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
|
|
|
|
|
Map<String, DeviceMessageHandler> handlerMap = applicationContext.getBeansOfType(DeviceMessageHandler.class);
|
|
|
|
|
messageHandlerPool = ThreadUtil.newScheduled(handlerMap.size() * 2, "messageHandler");
|
|
|
|
|
this.handlers.addAll(handlerMap.values());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@SneakyThrows
|
|
|
|
|
@Override
|
|
|
|
|
public void handler(ThingModelMessage msg) {
|
|
|
|
|
if (circuitOpenUntil > 0 && circuitOpenUntil > System.currentTimeMillis()) {
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
log.info("received thing model message:{}", msg);
|
|
|
|
|
try {
|
|
|
|
|
for (DeviceMessageHandler handler : this.handlers) {
|
|
|
|
|
@ -97,8 +93,17 @@ public class RuleDeviceConsumer implements ConsumerHandler<ThingModelMessage>, A
|
|
|
|
|
msg.setData(new HashMap<>());
|
|
|
|
|
}
|
|
|
|
|
handler.handle(msg);
|
|
|
|
|
// 成功 → 重置熔断
|
|
|
|
|
consecutiveFailures.set(0);
|
|
|
|
|
circuitOpenUntil = 0;
|
|
|
|
|
} catch (Throwable e) {
|
|
|
|
|
log.error("handler message error", e);
|
|
|
|
|
int failures = consecutiveFailures.incrementAndGet();
|
|
|
|
|
if (failures >= FAILURE_THRESHOLD) {
|
|
|
|
|
circuitOpenUntil = System.currentTimeMillis() + CIRCUIT_COOLDOWN_MS;
|
|
|
|
|
log.error("Circuit breaker OPEN for {}s, {} consecutive failures",
|
|
|
|
|
CIRCUIT_COOLDOWN_MS / 1000, failures);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
@ -107,4 +112,19 @@ public class RuleDeviceConsumer implements ConsumerHandler<ThingModelMessage>, A
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@PreDestroy
|
|
|
|
|
public void shutdown() {
|
|
|
|
|
if (messageHandlerPool != null) {
|
|
|
|
|
messageHandlerPool.shutdown();
|
|
|
|
|
try {
|
|
|
|
|
if (!messageHandlerPool.awaitTermination(10, TimeUnit.SECONDS)) {
|
|
|
|
|
messageHandlerPool.shutdownNow();
|
|
|
|
|
}
|
|
|
|
|
} catch (InterruptedException e) {
|
|
|
|
|
messageHandlerPool.shutdownNow();
|
|
|
|
|
Thread.currentThread().interrupt();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
