- Round-1 :① SHA256 hex 查表(写线程热点,5.9x);②SO_RCVBUF<=0 跳过显式设置;附随 JDK8 兼容补丁(生产级风险必修)

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