diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java
index cfcae0e..9625841 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java
@@ -36,9 +36,11 @@ public class AppProperties {
@Value("${syslog.tcp.backlog:4096}")
private int syslogTcpBacklog;
+ /** TCP SO_RCVBUF;<=0 时不显式设置(保留 OS 自适应与 Linux window scaling)。 */
@Value("${syslog.tcp.receive-buffer-bytes:4194304}")
private int syslogTcpReceiveBufferBytes;
+ /** UDP SO_RCVBUF;<=0 时不显式设置(保留 OS 自适应)。 */
@Value("${syslog.udp.receive-buffer-bytes:16777216}")
private int syslogUdpReceiveBufferBytes;
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/PendingWalAppend.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/PendingWalAppend.java
index 0ac7b85..0e943b2 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/PendingWalAppend.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/PendingWalAppend.java
@@ -2,17 +2,27 @@ package com.haobang.wal;
import java.util.concurrent.CompletableFuture;
-/** WAL 写线程队列中的事件及其待完成 durable future。 sequence/schemaVersion 由唯一写线程赋值,生产者线程不得提前推断 WAL 位置。 */
+/**
+ * WAL 写线程队列中的事件及其待完成 durable future。
+ *
+ *
sequence/schemaVersion 由唯一写线程赋值,生产者线程不得提前推断 WAL 位置。
+ * raw/rawSha256 由接入线程(admission)一次性预计算:它们与 sequence 无关,
+ * 前移后写线程只剩 POJO 组装、JSON 序列化与顺序 write。
+ */
final class PendingWalAppend {
private final IngressEvent event;
private final int rawBytes;
+ private final byte[] raw;
+ private final String rawSha256;
private final CompletableFuture durableFuture = new CompletableFuture<>();
private long sequence;
private int schemaVersion;
- PendingWalAppend(IngressEvent event, int rawBytes) {
+ PendingWalAppend(IngressEvent event, int rawBytes, byte[] raw, String rawSha256) {
this.event = event;
this.rawBytes = rawBytes;
+ this.raw = raw;
+ this.rawSha256 = rawSha256;
}
IngressEvent getEvent() {
@@ -21,6 +31,10 @@ final class PendingWalAppend {
int getRawBytes() { return rawBytes; }
+ byte[] getRaw() { return raw; }
+
+ String getRawSha256() { return rawSha256; }
+
CompletableFuture getDurableFuture() {
return durableFuture;
}
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalService.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalService.java
index baa76e0..c52369f 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalService.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalService.java
@@ -100,9 +100,15 @@ public class SyslogWalService {
private volatile long lastCompletedCleanupNanos;
private ArrayBlockingQueue appendQueue;
- /** 已 write、尚未被一次成功 force 覆盖的事件;仅由写线程访问。 */
+ /** 已 writeFully 成功、等待 force 的记录;由 pendingForce 锁保护(写线程追加,force 线程批量摘取)。 */
private final List pendingForce = new ArrayList<>();
private Thread writerThread;
+ /** 专职 force 线程:承担 FileChannel.force 与 checkpoint 刷盘,写线程只做顺序 write。 */
+ private Thread forceThread;
+ private final Object forceMonitor = new Object();
+ private final Object forceLock = new Object();
+ /** 写线程已出队、尚未 writeFully 成功登记的记录;failWal 时必须异常完成,否则 receipt 永久悬挂。 */
+ private volatile PendingWalAppend inFlightWrite;
private Path rootDirectory;
private Path activeDirectory;
private Path completedDirectory;
@@ -114,7 +120,8 @@ public class SyslogWalService {
private Path writerPath;
private LocalDate writerDate;
private long writerBytes;
- private int recordsSinceForce;
+ /** 仅在 pendingForce 锁内修改;volatile 仅为 forceDue() 的无锁读可见性。 */
+ private volatile int recordsSinceForce;
private long lastWrittenSequence;
private long lastForceNanos;
@@ -201,6 +208,9 @@ public class SyslogWalService {
writerThread = new Thread(this::writerLoop, "syslog-wal-writer");
writerThread.setDaemon(true);
writerThread.start();
+ forceThread = new Thread(this::forceLoop, "syslog-wal-forcer");
+ forceThread.setDaemon(true);
+ forceThread.start();
log.info("Syslog WAL 已启用: directory={}, charset={}, queueCapacity={}, highWatermark={}%, lowWatermark={}%, "
+ "retentionDays={}, completedCleanupIntervalMinutes={}, "
+ "diskUsedWatermark={}%, diskMaxFreeBytes={}, diskStopThresholdBytes={}, "
@@ -248,17 +258,25 @@ public class SyslogWalService {
}
return AppendAttempt.rejected(AdmissionStatus.UNAVAILABLE);
}
- int rawBytes = event.getRawMessage().getBytes(event.getCharset()).length;
+ // [SYSLOG-P1] 接入线程一次性完成 getBytes + SHA256(与 sequence 无关),写线程只剩 POJO/JSON/write。
+ byte[] raw = event.getRawMessage().getBytes(event.getCharset());
+ int rawBytes = raw.length;
if (properties.getSyslogMessageMaxBytes() > 0L
&& rawBytes > properties.getSyslogMessageMaxBytes()) {
admissionOversized.increment();
return AppendAttempt.rejected(AdmissionStatus.OVERSIZED);
}
+ String rawSha256;
+ try {
+ rawSha256 = sha256(raw);
+ } catch (Exception failure) {
+ throw new IllegalStateException("WAL 接入侧 SHA256 计算失败", failure);
+ }
if (!reservePendingBytes(rawBytes)) {
admissionBackpressure.increment();
return AppendAttempt.rejected(AdmissionStatus.BACKPRESSURE);
}
- PendingWalAppend pending = new PendingWalAppend(event, rawBytes);
+ PendingWalAppend pending = new PendingWalAppend(event, rawBytes, raw, rawSha256);
if (!appendQueue.offer(pending)) {
releasePendingBytes(rawBytes);
admissionBackpressure.increment();
@@ -434,16 +452,16 @@ public class SyslogWalService {
private void writerLoop() {
try {
+ // [SYSLOG-P1] force 与 checkpoint 刷盘均由专职 force 线程承担,写线程只做顺序 write,
+ // 消除 fsync 期间对 appendQueue 排空的阻塞。
while (running || !appendQueue.isEmpty()) {
PendingWalAppend pending = appendQueue.poll(
properties.getSyslogWalFlushIntervalMs(), TimeUnit.MILLISECONDS);
- if (pending != null) appendEvent(pending);
- long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastForceNanos);
- if (recordsSinceForce > 0 && (recordsSinceForce >= properties.getSyslogWalFlushRecords()
- || elapsedMs >= properties.getSyslogWalFlushIntervalMs())) {
- forceWriter();
+ if (pending != null) {
+ inFlightWrite = pending;
+ appendEvent(pending);
+ inFlightWrite = null;
}
- flushCheckpointIfDue();
}
forceWriter();
synchronized (checkpointMonitor) {
@@ -455,7 +473,11 @@ public class SyslogWalService {
} catch (Throwable failure) {
failWal(failure);
} finally {
- closeWriter();
+ inFlightWrite = null;
+ // 与 force 线程互斥后再关闭通道,避免 force 线程读到被置空的 writerChannel。
+ synchronized (forceLock) {
+ closeWriter();
+ }
}
}
@@ -463,9 +485,7 @@ public class SyslogWalService {
IngressEvent event = pending.getEvent();
long sequence = nextSequence.incrementAndGet();
pending.setSequence(sequence);
- // 当前元素已经离开队列;先登记,确保序列化/write 异常也会异常完成 receipt。
- pendingForce.add(pending);
- SyslogWalRecord record = toRecord(sequence, event);
+ SyslogWalRecord record = toRecord(sequence, event, pending.getRaw(), pending.getRawSha256());
pending.setSchemaVersion(record.getSchemaVersion());
byte[] json = objectMapper.writeValueAsBytes(record);
byte[] line = new byte[json.length + 1];
@@ -485,12 +505,22 @@ public class SyslogWalService {
}
writeFully(writerChannel, line);
writerBytes += line.length;
- recordsSinceForce++;
- lastWrittenSequence = sequence;
+ // 字节完全写入后才登记 force 等待:force 线程据此在 pendingForce 锁内快照 lastWrittenSequence,
+ // 保证 receipt 完成时对应字节必然已写入并随后被 force 覆盖,不出现"凭证超前落盘"。
+ synchronized (pendingForce) {
+ pendingForce.add(pending);
+ recordsSinceForce++;
+ lastWrittenSequence = sequence;
+ }
+ if (forceDue()) {
+ synchronized (forceMonitor) {
+ forceMonitor.notifyAll();
+ }
+ }
}
- private SyslogWalRecord toRecord(long sequence, IngressEvent event) throws Exception {
- byte[] raw = event.getRawMessage().getBytes(event.getCharset());
+ private SyslogWalRecord toRecord(long sequence, IngressEvent event, byte[] raw, String rawSha256)
+ throws Exception {
SyslogWalRecord record = new SyslogWalRecord();
if (event.getIngressType() == IngressType.API_PULL) {
record.setSchemaVersion(4);
@@ -517,7 +547,7 @@ public class SyslogWalService {
record.setProtocol(event.getTransport());
record.setCharset(event.getCharset().name());
record.setRawLength(raw.length);
- record.setRawSha256(sha256(raw));
+ record.setRawSha256(rawSha256);
record.setRawMessage(event.getRawMessage());
SyslogWalRecordValidator.validateRecord(record);
return record;
@@ -527,27 +557,81 @@ public class SyslogWalService {
return LocalDateTime.ofInstant(value, ZoneId.of("Asia/Shanghai")).format(RECORD_TIME);
}
- private void forceWriter() throws IOException {
- if (writerChannel == null || recordsSinceForce == 0) return;
- // receipt 代表可恢复边界,文件内容与长度元数据都必须先持久化。
- forceOperation.force(writerChannel, true);
- Instant durableAt = Instant.now();
- durableSequence.set(lastWrittenSequence);
- appended.add(recordsSinceForce);
- for (PendingWalAppend pending : pendingForce) {
- DurableReceipt receipt = new DurableReceipt(
- pending.getEvent().getEventId(),
- pending.getSequence(),
- pending.getSchemaVersion(),
- durableAt);
- releasePendingBytes(pending.getRawBytes());
- pending.getDurableFuture().complete(receipt);
+ /** 是否到达 force 触发条件(条数或时间);recordsSinceForce 为 volatile,读侧容忍轻微滞后。 */
+ private boolean forceDue() {
+ if (recordsSinceForce <= 0) return false;
+ if (recordsSinceForce >= properties.getSyslogWalFlushRecords()) return true;
+ return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastForceNanos)
+ >= properties.getSyslogWalFlushIntervalMs();
+ }
+
+ /** 专职 force 线程:fsync 与 checkpoint 刷盘移出写线程,写线程在此期间可继续排空 appendQueue。 */
+ private void forceLoop() {
+ try {
+ long tick = Math.min(properties.getSyslogWalCheckpointFlushIntervalMs(),
+ properties.getSyslogWalFlushIntervalMs());
+ while (running) {
+ synchronized (forceMonitor) {
+ forceMonitor.wait(tick);
+ }
+ if (!running) break;
+ if (forceDue()) forceWriter();
+ flushCheckpointIfDue();
+ }
+ // 兜底:writer 退出路径已做过最终 force,这里幂等补一次(pendingForce 为空则直接返回)。
+ forceWriter();
+ flushCheckpointIfDue();
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ } catch (Throwable failure) {
+ failWal(failure);
}
- pendingForce.clear();
- recordsSinceForce = 0;
- lastForceNanos = System.nanoTime();
- synchronized (durableMonitor) {
- durableMonitor.notifyAll();
+ }
+
+ private void forceWriter() throws IOException {
+ synchronized (forceLock) {
+ List batch;
+ long durableUpTo;
+ FileChannel channel;
+ synchronized (pendingForce) {
+ if (writerChannel == null || pendingForce.isEmpty()) return;
+ // 先摘批再 force:批内记录都在 writeFully 成功后才登记(pendingForce 锁保证 happens-before),
+ // 本次 force 必然覆盖整批字节;force 期间写线程的新写入会落入下一批。
+ // 写线程的段轮转(closeWriter/openWriter)同样要拿 forceLock,无法在 force 期间替换通道。
+ durableUpTo = lastWrittenSequence;
+ batch = new ArrayList<>(pendingForce);
+ pendingForce.clear();
+ recordsSinceForce = 0;
+ channel = writerChannel;
+ }
+ // receipt 代表可恢复边界,文件内容与长度元数据都必须先持久化。
+ try {
+ forceOperation.force(channel, true);
+ } catch (IOException forceFailure) {
+ // 批已摘除、尚未完成 receipt:force 失败必须显式异常完成整批并释放字节预算,
+ // 否则调用方永久悬挂(failWal 的兜底只覆盖 pendingForce/queue/inFlight)。
+ for (PendingWalAppend pending : batch) {
+ releasePendingBytes(pending.getRawBytes());
+ pending.getDurableFuture().completeExceptionally(forceFailure);
+ }
+ throw forceFailure;
+ }
+ Instant durableAt = Instant.now();
+ durableSequence.set(durableUpTo);
+ appended.add(batch.size());
+ for (PendingWalAppend pending : batch) {
+ DurableReceipt receipt = new DurableReceipt(
+ pending.getEvent().getEventId(),
+ pending.getSequence(),
+ pending.getSchemaVersion(),
+ durableAt);
+ releasePendingBytes(pending.getRawBytes());
+ pending.getDurableFuture().complete(receipt);
+ }
+ lastForceNanos = System.nanoTime();
+ synchronized (durableMonitor) {
+ durableMonitor.notifyAll();
+ }
}
}
@@ -574,20 +658,34 @@ public class SyslogWalService {
private void failWal(Throwable failure) {
fatalFailure = failure;
running = false;
+ // 已出队但尚未 writeFully 成功登记的记录不在 pendingForce/queue 中,必须单独兜底。
+ PendingWalAppend inflight = inFlightWrite;
+ if (inflight != null) {
+ inFlightWrite = null;
+ releasePendingBytes(inflight.getRawBytes());
+ inflight.getDurableFuture().completeExceptionally(failure);
+ }
// 任何尚未 durable 的调用方都必须收到失败,不能留下永久等待或假成功的 receipt。
completePendingExceptionally(failure);
synchronized (durableMonitor) {
durableMonitor.notifyAll();
}
+ synchronized (forceMonitor) {
+ forceMonitor.notifyAll();
+ }
log.error("Syslog WAL 发生致命故障,接入已进入 fail-closed 状态", failure);
}
private void completePendingExceptionally(Throwable failure) {
- for (PendingWalAppend pending : pendingForce) {
+ List batch;
+ synchronized (pendingForce) {
+ batch = new ArrayList<>(pendingForce);
+ pendingForce.clear();
+ }
+ for (PendingWalAppend pending : batch) {
releasePendingBytes(pending.getRawBytes());
pending.getDurableFuture().completeExceptionally(failure);
}
- pendingForce.clear();
if (appendQueue == null) return;
PendingWalAppend pending;
while ((pending = appendQueue.poll()) != null) {
@@ -698,14 +796,15 @@ public class SyslogWalService {
ByteBuffer one = ByteBuffer.allocate(1);
channel.position(size - 1);
channel.read(one);
- one.flip();
+ // [SYSLOG-COMPAT] JDK9+ 的 ByteBuffer 协变返回类型在 JDK8 运行时会 NoSuchMethodError,统一强转 Buffer
+ ((java.nio.Buffer) one).flip();
if (one.get() == (byte) '\n') continue;
long position = size - 1;
while (position >= 0) {
- one.clear();
+ ((java.nio.Buffer) one).clear();
channel.position(position);
channel.read(one);
- one.flip();
+ ((java.nio.Buffer) one).flip();
if (one.get() == (byte) '\n') break;
position--;
}
@@ -885,10 +984,16 @@ public class SyslogWalService {
return stopThresholdBytes > 0L && usableBytes <= stopThresholdBytes;
}
+ private static final char[] HEX_DIGITS = "0123456789abcdef".toCharArray();
+
private String sha256(byte[] value) throws Exception {
byte[] digest = MessageDigest.getInstance("SHA-256").digest(value);
StringBuilder hex = new StringBuilder(digest.length * 2);
- for (byte item : digest) hex.append(String.format("%02x", item & 0xff));
+ // 查表替换 String.format("%02x"):writer 线程上 Formatter 正则解析是可测热点
+ for (byte item : digest) {
+ int v = item & 0xff;
+ hex.append(HEX_DIGITS[v >>> 4]).append(HEX_DIGITS[v & 0x0f]);
+ }
return hex.toString();
}
@@ -932,6 +1037,7 @@ public class SyslogWalService {
public void shutdown() {
running = false;
synchronized (durableMonitor) { durableMonitor.notifyAll(); }
+ synchronized (forceMonitor) { forceMonitor.notifyAll(); }
if (writerThread != null) {
try {
writerThread.join(properties.getSyslogWalShutdownTimeoutMs());
@@ -944,13 +1050,28 @@ public class SyslogWalService {
Thread.currentThread().interrupt();
}
}
+ if (forceThread != null) {
+ try {
+ forceThread.join(Math.min(1000L, properties.getSyslogWalShutdownTimeoutMs()));
+ if (forceThread.isAlive()) forceThread.interrupt();
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
synchronized (checkpointMonitor) { flushCheckpointLocked(); }
}
public final class WalCursor implements AutoCloseable {
+ private static final int READ_BUF_SIZE = 1 << 16;
private String currentSegment;
private long position;
private FileChannel channel;
+ // [SYSLOG-P1] 持久化读缓冲:块内多行只读一次磁盘,替代旧实现"每行重新 position+read 8KB"。
+ private ByteBuffer readBuf;
+ private long readBufBase;
+ private int readBufPos;
+ private int readBufLen;
+ private boolean needSeek = true;
private WalCursor(Checkpoint checkpoint) throws IOException {
this.currentSegment = checkpoint.segment;
@@ -983,20 +1104,20 @@ public class SyslogWalService {
}
SyslogWalRecord record;
try {
- JsonNode node = objectMapper.readTree(line);
+ // [SYSLOG-P1] 单次解析替代旧 readTree+treeToValue 双解析;仅损坏记录才回退到树解析做墓碑。
+ record = objectMapper.readValue(line, SyslogWalRecord.class);
+ SyslogWalRecordValidator.validateRecord(record);
+ } catch (Exception invalidRecord) {
try {
- SyslogWalRecordValidator.validateNode(node);
- record = objectMapper.treeToValue(node, SyslogWalRecord.class);
- SyslogWalRecordValidator.validateRecord(record);
- } catch (Exception invalidRecord) {
- record = invalidRecordTombstone(node, invalidRecord);
+ record = invalidRecordTombstone(objectMapper.readTree(line), invalidRecord);
+ } catch (Exception failure) {
+ throw new IOException("WAL 中间记录损坏,拒绝越过: segment=" + currentSegment
+ + ", offset=" + start, failure);
}
- } catch (Exception failure) {
- throw new IOException("WAL 中间记录损坏,拒绝越过: segment=" + currentSegment
- + ", offset=" + start, failure);
}
if (record.getSequence() > durableSequence.get()) {
position = start;
+ invalidateBuffer();
awaitDurable(deadline);
if (System.nanoTime() >= deadline) return null;
continue;
@@ -1037,6 +1158,7 @@ public class SyslogWalService {
currentSegment = first.getFileName().toString();
position = 0L;
channel = FileChannel.open(first, StandardOpenOption.READ);
+ invalidateBuffer();
}
private boolean openNextSegment() throws IOException {
@@ -1048,36 +1170,85 @@ public class SyslogWalService {
currentSegment = name;
position = 0L;
channel = FileChannel.open(path, StandardOpenOption.READ);
+ invalidateBuffer();
return true;
}
}
return false;
}
+ /** 丢弃读缓冲:position 被回退(等待 durable)或通道切换后必须重定位。 */
+ private void invalidateBuffer() {
+ readBufPos = 0;
+ readBufLen = 0;
+ needSeek = true;
+ }
+
+ /** 从磁盘补一段进读缓冲;EOF 返回 false。 */
+ private boolean fillBuffer() throws IOException {
+ if (readBuf == null) readBuf = ByteBuffer.allocate(READ_BUF_SIZE);
+ if (needSeek) {
+ readBufBase = position;
+ readBufPos = 0;
+ readBufLen = 0;
+ channel.position(position);
+ needSeek = false;
+ }
+ byte[] arr = readBuf.array();
+ if (readBufPos > 0) {
+ int tail = readBufLen - readBufPos;
+ if (tail > 0) System.arraycopy(arr, readBufPos, arr, 0, tail);
+ readBufBase += readBufPos;
+ readBufLen = tail;
+ readBufPos = 0;
+ }
+ if (readBufLen == arr.length) return false;
+ ((java.nio.Buffer) readBuf).clear();
+ ((java.nio.Buffer) readBuf).position(readBufLen);
+ int n = channel.read(readBuf);
+ if (n <= 0) return false;
+ readBufLen += n;
+ return readBufLen > readBufPos;
+ }
+
private byte[] readLine() throws IOException {
long start = position;
- ByteArrayOutputStream output = new ByteArrayOutputStream(2048);
- ByteBuffer buffer = ByteBuffer.allocate(8192);
+ ByteArrayOutputStream overflow = null;
while (true) {
- buffer.clear();
- channel.position(position);
- int count = channel.read(buffer);
- if (count < 0 || count == 0) {
+ if (readBufPos >= readBufLen && !fillBuffer()) {
+ // EOF:半行丢弃并回退(与旧实现一致,未完成尾行由启动期 repair 处理)。
position = start;
return null;
}
- buffer.flip();
- int consumed = 0;
- while (buffer.hasRemaining()) {
- byte value = buffer.get();
- consumed++;
- if (value == (byte) '\n') {
- position += consumed;
- return output.toByteArray();
+ byte[] arr = readBuf.array();
+ for (int i = readBufPos; i < readBufLen; i++) {
+ if (arr[i] == (byte) '\n') {
+ int chunk = i - readBufPos;
+ byte[] line;
+ if (overflow == null) {
+ line = new byte[chunk];
+ System.arraycopy(arr, readBufPos, line, 0, chunk);
+ } else {
+ overflow.write(arr, readBufPos, chunk);
+ line = overflow.toByteArray();
+ }
+ readBufPos = i + 1;
+ position = readBufBase + readBufPos;
+ return line;
}
- output.write(value);
}
- position += consumed;
+ // 本块无换行:整体并入溢出流继续读
+ if (overflow == null) {
+ overflow = new ByteArrayOutputStream(Math.max(4096, (readBufLen - readBufPos) * 2));
+ }
+ overflow.write(arr, readBufPos, readBufLen - readBufPos);
+ if (overflow.size() > 32_000_000) {
+ position = start;
+ throw new IOException("WAL 行超长(32MB 内无换行): segment=" + currentSegment
+ + ", offset=" + start);
+ }
+ readBufPos = readBufLen;
+ position = readBufBase + readBufPos;
}
}
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerBoth.java b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerBoth.java
index c6c1ac5..69212b6 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerBoth.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerBoth.java
@@ -126,8 +126,12 @@ public class SyslogServerBoth {
// [SYSLOG-ALARM][NINGRESS][已修复] backlog/接收缓冲配置化;实际生效上限仍受操作系统 somaxconn/rmem_max 约束。
.option(ChannelOption.SO_BACKLOG, properties.getSyslogTcpBacklog())
.childOption(ChannelOption.SO_KEEPALIVE, true)
- .childOption(ChannelOption.AUTO_READ, true)
- .childOption(ChannelOption.SO_RCVBUF, properties.getSyslogTcpReceiveBufferBytes());
+ .childOption(ChannelOption.AUTO_READ, true);
+ // [SYSLOG-VERIFY] SO_RCVBUF<=0 时不显式设置:显式设置会禁用 Linux window scaling(窗口锁死 64KB)
+ // 以及 Windows 接收窗口自动调优,实测会导致突发流量下窗口塌缩与吞吐骤降。
+ if (properties.getSyslogTcpReceiveBufferBytes() > 0) {
+ b.childOption(ChannelOption.SO_RCVBUF, properties.getSyslogTcpReceiveBufferBytes());
+ }
tcpChannelFuture = b.bind(tcpPort).sync();
logger.info("TCP Syslog server started on port {}", tcpPort);
@@ -147,8 +151,11 @@ public class SyslogServerBoth {
.handler(new SyslogUdpChannelInitializer(syslogCharset, walService, deviceResolver,
!properties.isSyslogDispatchTerminalRecordEnabled(), ipMetrics))
// [SYSLOG-ALARM][NINGRESS][已修复] 显式配置 UDP 接收缓冲;UDP 仍无端到端确认,强可靠设备应改用 TCP/TLS 或 RELP。
- .option(ChannelOption.SO_BROADCAST, true)
- .option(ChannelOption.SO_RCVBUF, properties.getSyslogUdpReceiveBufferBytes());
+ .option(ChannelOption.SO_BROADCAST, true);
+ // [SYSLOG-VERIFY] 与 TCP 一致:<=0 表示交给操作系统自适应。
+ if (properties.getSyslogUdpReceiveBufferBytes() > 0) {
+ b.option(ChannelOption.SO_RCVBUF, properties.getSyslogUdpReceiveBufferBytes());
+ }
udpChannelFuture = b.bind(udpPort).sync();
logger.info("UDP Syslog server started on port {}", udpPort);