266 lines
12 KiB
Java
266 lines
12 KiB
Java
|
package com.haobang.dispatch;
|
||
|
|
|
||
|
|
import com.haobang.config.AppProperties;
|
||
|
|
import com.haobang.wal.SyslogWalRecord;
|
||
|
|
import com.haobang.wal.SyslogWalService;
|
||
|
|
import com.kafka.kafkaProducer;
|
||
|
|
import com.netty.SyslogMessageHandler;
|
||
|
|
import org.slf4j.Logger;
|
||
|
|
import org.slf4j.LoggerFactory;
|
||
|
|
import org.springframework.stereotype.Component;
|
||
|
|
|
||
|
|
import javax.annotation.PostConstruct;
|
||
|
|
import javax.annotation.PreDestroy;
|
||
|
|
import java.time.LocalDateTime;
|
||
|
|
import java.time.format.DateTimeFormatter;
|
||
|
|
import java.util.HashMap;
|
||
|
|
import java.util.Map;
|
||
|
|
import java.util.concurrent.ExecutorService;
|
||
|
|
import java.util.concurrent.Executors;
|
||
|
|
import java.util.concurrent.ScheduledExecutorService;
|
||
|
|
import java.util.concurrent.Semaphore;
|
||
|
|
import java.util.concurrent.ThreadFactory;
|
||
|
|
import java.util.concurrent.TimeUnit;
|
||
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||
|
|
import java.util.concurrent.atomic.AtomicLong;
|
||
|
|
import java.util.concurrent.atomic.LongAdder;
|
||
|
|
|
||
|
|
/** 从 WAL 有界并发异步投递 Kafka，并只按连续 ACK 推进 WAL 游标。 */
|
||
|
|
@Component
|
||
|
|
public class SyslogWalDispatcher {
|
||
|
|
private static final Logger log = LoggerFactory.getLogger(SyslogWalDispatcher.class);
|
||
|
|
private static final DateTimeFormatter WAL_TIME = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS");
|
||
|
|
private static final DateTimeFormatter HEADER_TIME = DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS");
|
||
|
|
|
||
|
|
private final SyslogWalService walService;
|
||
|
|
private final DeviceContextResolver deviceResolver;
|
||
|
|
private final kafkaProducer producer;
|
||
|
|
private final AppProperties properties;
|
||
|
|
private final AtomicBoolean running = new AtomicBoolean();
|
||
|
|
private final Object completionMonitor = new Object();
|
||
|
|
private final Map<Long, SyslogWalRecord> completedOutOfOrder = new HashMap<>();
|
||
|
|
private final LongAdder kafkaAcked = new LongAdder();
|
||
|
|
private final LongAdder terminalFiltered = new LongAdder();
|
||
|
|
private final LongAdder retryCount = new LongAdder();
|
||
|
|
private final AtomicLong inflight = new AtomicLong();
|
||
|
|
|
||
|
|
private Semaphore permits;
|
||
|
|
private ExecutorService workers;
|
||
|
|
private ScheduledExecutorService scheduler;
|
||
|
|
private Thread readerThread;
|
||
|
|
private SyslogWalService.WalCursor cursor;
|
||
|
|
private long nextCommitSequence;
|
||
|
|
private long nextReadSequence;
|
||
|
|
|
||
|
|
public SyslogWalDispatcher(SyslogWalService walService,
|
||
|
|
DeviceContextResolver deviceResolver,
|
||
|
|
kafkaProducer producer,
|
||
|
|
AppProperties properties) {
|
||
|
|
this.walService = walService;
|
||
|
|
this.deviceResolver = deviceResolver;
|
||
|
|
this.producer = producer;
|
||
|
|
this.properties = properties;
|
||
|
|
}
|
||
|
|
|
||
|
|
@PostConstruct
|
||
|
|
public void start() throws Exception {
|
||
|
|
validateConfiguration();
|
||
|
|
permits = new Semaphore(properties.getSyslogDispatchMaxInflight());
|
||
|
|
workers = Executors.newFixedThreadPool(properties.getSyslogDispatchWorkerThreads(),
|
||
|
|
namedFactory("syslog-dispatch-worker"));
|
||
|
|
scheduler = Executors.newScheduledThreadPool(2, namedFactory("syslog-dispatch-scheduler"));
|
||
|
|
cursor = walService.openCursor();
|
||
|
|
nextCommitSequence = walService.getCheckpointSequence() + 1L;
|
||
|
|
nextReadSequence = nextCommitSequence;
|
||
|
|
running.set(true);
|
||
|
|
readerThread = new Thread(this::readLoop, "syslog-wal-dispatch-reader");
|
||
|
|
readerThread.setDaemon(true);
|
||
|
|
readerThread.start();
|
||
|
|
scheduler.scheduleWithFixedDelay(this::logStatus,
|
||
|
|
properties.getSyslogStatusLogIntervalMs(), properties.getSyslogStatusLogIntervalMs(),
|
||
|
|
TimeUnit.MILLISECONDS);
|
||
|
|
log.info("Syslog WAL Dispatcher 已启动: workerThreads={}, maxInflight={}, checkpointSequence={}",
|
||
|
|
properties.getSyslogDispatchWorkerThreads(), properties.getSyslogDispatchMaxInflight(),
|
||
|
|
walService.getCheckpointSequence());
|
||
|
|
}
|
||
|
|
|
||
|
|
private void readLoop() {
|
||
|
|
try {
|
||
|
|
while (running.get()) {
|
||
|
|
permits.acquire();
|
||
|
|
SyslogWalRecord record = cursor.next(1000L);
|
||
|
|
if (record == null) {
|
||
|
|
permits.release();
|
||
|
|
continue;
|
||
|
|
}
|
||
|
|
if (record.getSequence() != nextReadSequence) {
|
||
|
|
permits.release();
|
||
|
|
throw new IllegalStateException("WAL sequence 不连续，拒绝越过: expected="
|
||
|
|
+ nextReadSequence + ", actual=" + record.getSequence());
|
||
|
|
}
|
||
|
|
nextReadSequence++;
|
||
|
|
inflight.incrementAndGet();
|
||
|
|
workers.execute(() -> dispatch(record, 1));
|
||
|
|
}
|
||
|
|
} catch (InterruptedException interrupted) {
|
||
|
|
Thread.currentThread().interrupt();
|
||
|
|
} catch (Throwable failure) {
|
||
|
|
running.set(false);
|
||
|
|
log.error("WAL Dispatcher 读取失败，停止推进游标以保护数据", failure);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
private void dispatch(SyslogWalRecord record, int attempt) {
|
||
|
|
if (!running.get()) return;
|
||
|
|
if (record.isQuarantinedInvalid()) {
|
||
|
|
log.error("损坏 WAL 已隔离并推进游标: sequence={}, eventId={}, reason={}",
|
||
|
|
record.getSequence(), record.getEventId(), record.getValidationError());
|
||
|
|
markComplete(record);
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
final DeviceContextResolver.Resolution resolution;
|
||
|
|
try {
|
||
|
|
resolution = deviceResolver.resolve(record.getSourceIp(), record.getProtocol());
|
||
|
|
} catch (Throwable failure) {
|
||
|
|
scheduleRetry(record, attempt, "DEVICE_LOOKUP", failure);
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
|
||
|
|
if (!resolution.isRegistered()) {
|
||
|
|
try {
|
||
|
|
walService.recordTerminal(record, resolution.getType().name(), resolution.getReason());
|
||
|
|
terminalFiltered.increment();
|
||
|
|
markComplete(record);
|
||
|
|
} catch (Throwable failure) {
|
||
|
|
scheduleRetry(record, attempt, "TERMINAL_LEDGER", failure);
|
||
|
|
}
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
|
||
|
|
final String receiveTime;
|
||
|
|
try {
|
||
|
|
receiveTime = LocalDateTime.parse(record.getReceiveTime(), WAL_TIME).format(HEADER_TIME);
|
||
|
|
producer.messagePush(resolution.getDevice(), record.getRawMessage(), receiveTime,
|
||
|
|
record.getEventId(), record.getPointTimeNs())
|
||
|
|
.whenComplete((delivery, failure) -> {
|
||
|
|
if (failure == null) {
|
||
|
|
kafkaAcked.increment();
|
||
|
|
markComplete(record);
|
||
|
|
} else {
|
||
|
|
scheduleRetry(record, attempt, "KAFKA_ACK", unwrap(failure));
|
||
|
|
}
|
||
|
|
});
|
||
|
|
} catch (Throwable failure) {
|
||
|
|
// 加密、配置或 Producer 本地错误都不能越过 WAL；修复依赖后沿用同一 event_id 重试。
|
||
|
|
scheduleRetry(record, attempt, "KAFKA_PREPARE", failure);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
private void scheduleRetry(SyslogWalRecord record, int attempt, String stage, Throwable failure) {
|
||
|
|
retryCount.increment();
|
||
|
|
long retries = retryCount.sum();
|
||
|
|
if (attempt == 1 || retries % 100 == 0) {
|
||
|
|
log.warn("WAL 事件投递失败，将保留原 event_id 重试: eventId={}, sequence={}, stage={}, "
|
||
|
|
+ "attempt={}, retryTotal={}, error={}",
|
||
|
|
record.getEventId(), record.getSequence(), stage, attempt, retries,
|
||
|
|
failure == null ? "unknown" : failure.toString());
|
||
|
|
}
|
||
|
|
if (!running.get()) return;
|
||
|
|
scheduler.schedule(() -> {
|
||
|
|
if (running.get()) workers.execute(() -> dispatch(record, attempt + 1));
|
||
|
|
}, properties.getSyslogDispatchRetryBackoffMs(), TimeUnit.MILLISECONDS);
|
||
|
|
}
|
||
|
|
|
||
|
|
private void markComplete(SyslogWalRecord record) {
|
||
|
|
synchronized (completionMonitor) {
|
||
|
|
if (record.getSequence() < nextCommitSequence) return;
|
||
|
|
completedOutOfOrder.put(record.getSequence(), record);
|
||
|
|
SyslogWalRecord latest = null;
|
||
|
|
int released = 0;
|
||
|
|
while (true) {
|
||
|
|
SyslogWalRecord contiguous = completedOutOfOrder.remove(nextCommitSequence);
|
||
|
|
if (contiguous == null) break;
|
||
|
|
latest = contiguous;
|
||
|
|
nextCommitSequence++;
|
||
|
|
released++;
|
||
|
|
}
|
||
|
|
if (latest != null) walService.markCheckpoint(latest);
|
||
|
|
if (released > 0) {
|
||
|
|
inflight.addAndGet(-released);
|
||
|
|
permits.release(released);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
private void logStatus() {
|
||
|
|
if (!running.get()) return;
|
||
|
|
log.info("Syslog WAL状态: networkReceived={}, appended={}, walPendingApprox={}, walQueue={}/{}, durableSequence={}, "
|
||
|
|
+ "checkpointSequence={}, kafkaInflight={}, kafkaAcked={}, terminalFiltered={}, "
|
||
|
|
+ "retry={}, tcpPauseTotal={}, udpDropped={}, deviceCacheSize={}, deviceCacheHits={}, "
|
||
|
|
+ "deviceCacheMisses={}",
|
||
|
|
SyslogMessageHandler.getNetworkReceivedCount(), walService.getAppendedCount(),
|
||
|
|
walService.getPendingRecordsApprox(),
|
||
|
|
walService.getQueueSize(), walService.getQueueCapacity(), walService.getDurableSequence(),
|
||
|
|
walService.getCheckpointSequence(), inflight.get(), kafkaAcked.sum(), terminalFiltered.sum(),
|
||
|
|
retryCount.sum(), SyslogMessageHandler.getTcpPauseCount(),
|
||
|
|
SyslogMessageHandler.getUdpDroppedCount(), deviceResolver.getCacheSize(),
|
||
|
|
deviceResolver.getHitCount(), deviceResolver.getMissCount());
|
||
|
|
}
|
||
|
|
|
||
|
|
private Throwable unwrap(Throwable failure) {
|
||
|
|
if (failure instanceof java.util.concurrent.CompletionException && failure.getCause() != null) {
|
||
|
|
return failure.getCause();
|
||
|
|
}
|
||
|
|
return failure;
|
||
|
|
}
|
||
|
|
|
||
|
|
private ThreadFactory namedFactory(String prefix) {
|
||
|
|
AtomicLong sequence = new AtomicLong();
|
||
|
|
return runnable -> {
|
||
|
|
Thread thread = new Thread(runnable, prefix + "-" + sequence.incrementAndGet());
|
||
|
|
thread.setDaemon(true);
|
||
|
|
return thread;
|
||
|
|
};
|
||
|
|
}
|
||
|
|
|
||
|
|
private void validateConfiguration() {
|
||
|
|
if (properties.getSyslogDispatchWorkerThreads() <= 0
|
||
|
|
|| properties.getSyslogDispatchMaxInflight() <= 0
|
||
|
|
|| properties.getSyslogDispatchRetryBackoffMs() <= 0
|
||
|
|
|| properties.getSyslogDispatchShutdownTimeoutMs() <= 0
|
||
|
|
|| properties.getSyslogDeviceContextCacheTtlMs() <= 0
|
||
|
|
|| properties.getSyslogStatusLogIntervalMs() <= 0) {
|
||
|
|
throw new IllegalArgumentException("Dispatcher 线程、在途、重试、缓存和状态日志参数必须大于 0");
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
public long getInflight() { return inflight.get(); }
|
||
|
|
public long getKafkaAcked() { return kafkaAcked.sum(); }
|
||
|
|
public long getTerminalFiltered() { return terminalFiltered.sum(); }
|
||
|
|
public long getRetryCount() { return retryCount.sum(); }
|
||
|
|
|
||
|
|
@PreDestroy
|
||
|
|
public void shutdown() {
|
||
|
|
if (!running.compareAndSet(true, false)) return;
|
||
|
|
if (readerThread != null) readerThread.interrupt();
|
||
|
|
if (scheduler != null) scheduler.shutdownNow();
|
||
|
|
if (workers != null) workers.shutdown();
|
||
|
|
long deadline = System.nanoTime()
|
||
|
|
+ TimeUnit.MILLISECONDS.toNanos(properties.getSyslogDispatchShutdownTimeoutMs());
|
||
|
|
try {
|
||
|
|
while (inflight.get() > 0 && System.nanoTime() < deadline) Thread.sleep(20L);
|
||
|
|
if (workers != null) {
|
||
|
|
long remaining = Math.max(1L, deadline - System.nanoTime());
|
||
|
|
workers.awaitTermination(remaining, TimeUnit.NANOSECONDS);
|
||
|
|
}
|
||
|
|
} catch (InterruptedException interrupted) {
|
||
|
|
Thread.currentThread().interrupt();
|
||
|
|
}
|
||
|
|
if (workers != null && !workers.isTerminated()) workers.shutdownNow();
|
||
|
|
try { if (cursor != null) cursor.close(); }
|
||
|
|
catch (Exception failure) { log.warn("关闭 WAL cursor 失败", failure); }
|
||
|
|
log.info("Syslog WAL Dispatcher 已停止: inflight={}, checkpointSequence={}",
|
||
|
|
inflight.get(), walService.getCheckpointSequence());
|
||
|
|
}
|
||
|
|
}
|