From 100fad7d18565127b2b28243fa07f63476b0bd3f Mon Sep 17 00:00:00 2001
From: KNEMC <2014539504@qq.com>
Date: Tue, 29 Sep 2026 10:48:49 +0800
Subject: [PATCH] cmt
---
.idea/vcs.xml | 1 +
.../network/AsynchronousDatagramChannel.java | 277 ++++++++++++++++++
.../cloud/network/klalb/KLALBController.java | 4 +-
3 files changed, 280 insertions(+), 2 deletions(-)
create mode 100644 src/org/kne/cloud/network/AsynchronousDatagramChannel.java
diff --git a/.idea/vcs.xml b/.idea/vcs.xml
index 35eb1dd..8f85844 100644
--- a/.idea/vcs.xml
+++ b/.idea/vcs.xml
@@ -2,5 +2,6 @@
+
\ No newline at end of file
diff --git a/src/org/kne/cloud/network/AsynchronousDatagramChannel.java b/src/org/kne/cloud/network/AsynchronousDatagramChannel.java
new file mode 100644
index 0000000..c96deb9
--- /dev/null
+++ b/src/org/kne/cloud/network/AsynchronousDatagramChannel.java
@@ -0,0 +1,277 @@
+package org.kne.cloud.network;
+
+import java.io.IOException;
+import java.net.DatagramSocket;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.NetworkInterface;
+import java.net.SocketAddress;
+import java.net.SocketOption;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.DatagramChannel;
+import java.nio.channels.MembershipKey;
+import java.nio.channels.SelectableChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.spi.SelectorProvider;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * 自制的 UDP AIO 通道，内部使用线程池模拟异步收包。
+ * 用户只需设置回调，无需自己管理线程。
+ *
+ * 完整委托了 DatagramChannel 的所有公共方法。
+ */
+public class AsynchronousDatagramChannel implements AutoCloseable {
+
+ private final DatagramChannel delegate;
+ private final ExecutorService workerPool;
+ private final AtomicBoolean running = new AtomicBoolean(true);
+ private volatile ReceiveHandler handler;
+ private final int bufferSize;
+ private int threadCount;
+
+ // ==================== 回调接口 ====================
+
+ @FunctionalInterface
+ public interface ReceiveHandler {
+ void onReceive(SocketAddress from, ByteBuffer data);
+ }
+
+ // ==================== 构造方法 ====================
+
+ public AsynchronousDatagramChannel() throws IOException {
+ this(65535, Runtime.getRuntime().availableProcessors());
+ }
+
+ public AsynchronousDatagramChannel(int bufferSize, int threadCount) throws IOException {
+ this.bufferSize = bufferSize;
+ this.delegate = DatagramChannel.open();
+ delegate.configureBlocking(true);
+ this.threadCount=threadCount;
+ this.workerPool = Executors.newFixedThreadPool(threadCount, r -> {
+ Thread t = new Thread(r, "AsyncDatagramReader");
+ t.setDaemon(true);
+ return t;
+ });
+
+ }
+
+ public void startWorkers() {
+
+ for (int i = 0; i < threadCount; i++) {
+ workerPool.submit(this::readerLoop);
+ }
+ }
+
+ // ==================== 核心方法 ====================
+
+ public void setReceiveHandler(ReceiveHandler handler) {
+ this.handler = handler;
+ }
+
+ private void readerLoop() {
+ ByteBuffer buffer = ByteBuffer.allocateDirect(bufferSize);
+ while (running.get()) {
+ try {
+ buffer.clear();
+ SocketAddress from = delegate.receive(buffer);
+ if (from == null) {
+ continue;
+ }
+ buffer.flip();
+ ReceiveHandler localHandler = handler;
+ if (localHandler != null) {
+ localHandler.onReceive(from, buffer);
+ }
+ } catch (IOException e) {
+ if (running.get()) {
+ e.printStackTrace();
+ }
+ break;
+ }
+ }
+ }
+
+ public int send(ByteBuffer data, SocketAddress target) throws IOException {
+ return delegate.send(data, target);
+ }
+
+ // ==================== DatagramChannel 委托方法 ====================
+
+ // --- 连接相关 ---
+
+ public DatagramChannel connect(SocketAddress remote) throws IOException {
+ return delegate.connect(remote);
+ }
+
+ public DatagramChannel disconnect() throws IOException {
+ return delegate.disconnect();
+ }
+
+ public boolean isConnected() {
+ return delegate.isConnected();
+ }
+
+ // --- 绑定相关 ---
+
+ public DatagramChannel bind(SocketAddress local) throws IOException {
+ return delegate.bind(local);
+ }
+
+ public SocketAddress getLocalAddress() throws IOException {
+ return delegate.getLocalAddress();
+ }
+
+ public SocketAddress getRemoteAddress() throws IOException {
+ return delegate.getRemoteAddress();
+ }
+
+ // --- 读写相关 ---
+
+ public int read(ByteBuffer dst) throws IOException {
+ return delegate.read(dst);
+ }
+
+ public long read(ByteBuffer[] dsts) throws IOException {
+ return delegate.read(dsts);
+ }
+
+ public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
+ return delegate.read(dsts, offset, length);
+ }
+
+ public int write(ByteBuffer src) throws IOException {
+ return delegate.write(src);
+ }
+
+ public long write(ByteBuffer[] srcs) throws IOException {
+ return delegate.write(srcs);
+ }
+
+ public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+ return delegate.write(srcs, offset, length);
+ }
+
+ // --- 接收相关 ---
+
+ public SocketAddress receive(ByteBuffer dst) throws IOException {
+ return delegate.receive(dst);
+ }
+
+ // --- 多播 (Multicast) ---
+
+ public MembershipKey join(InetAddress group, NetworkInterface interf) throws IOException {
+ return delegate.join(group, interf);
+ }
+
+ public MembershipKey join(InetAddress group, NetworkInterface interf, InetAddress source) throws IOException {
+ return delegate.join(group, interf, source);
+ }
+
+ // --- 选项相关 ---
+
+ public DatagramChannel setOption(SocketOption name, T value) throws IOException {
+ return delegate.setOption(name, value);
+ }
+
+ public T getOption(SocketOption name) throws IOException {
+ return delegate.getOption(name);
+ }
+
+ public Set> supportedOptions() {
+ return delegate.supportedOptions();
+ }
+
+ // --- 阻塞模式 ---
+
+
+ public boolean isBlocking() {
+ return delegate.isBlocking();
+ }
+
+ // --- 选择器相关 ---
+
+ public SelectableChannel configureBlocking(boolean block) throws IOException {
+ return delegate.configureBlocking(block);
+ }
+
+ public SelectorProvider provider() {
+ return delegate.provider();
+ }
+
+ public int validOps() {
+ return delegate.validOps();
+ }
+
+ public SelectionKey keyFor(Selector sel) {
+ return delegate.keyFor(sel);
+ }
+
+ public SelectionKey register(Selector sel, int ops) throws ClosedChannelException {
+ return delegate.register(sel, ops);
+ }
+
+ public SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException {
+ return delegate.register(sel, ops, att);
+ }
+
+ // --- 状态查询 ---
+
+ public boolean isOpen() {
+ return delegate.isOpen() && running.get();
+ }
+
+ public DatagramSocket socket() {
+ return delegate.socket();
+ }
+
+ // ==================== 关闭方法 ====================
+
+ @Override
+ public void close() throws Exception {
+ running.set(false);
+ workerPool.shutdownNow();
+ delegate.close();
+ }
+
+ // ==================== 静态工厂方法 ====================
+
+ public static AsynchronousDatagramChannel open() throws IOException {
+ return new AsynchronousDatagramChannel();
+ }
+
+ public static AsynchronousDatagramChannel open(int bufferSize) throws IOException {
+ return new AsynchronousDatagramChannel(bufferSize, Runtime.getRuntime().availableProcessors());
+ }
+
+ public static AsynchronousDatagramChannel open(int bufferSize, int threadCount) throws IOException {
+ return new AsynchronousDatagramChannel(bufferSize, threadCount);
+ }
+
+ // ==================== 测试示例 ====================
+
+ public static void main(String[] args) throws Exception {
+ int port = 4567;
+ AsynchronousDatagramChannel channel = AsynchronousDatagramChannel.open();
+ channel.bind(new InetSocketAddress(port));
+
+ channel.setReceiveHandler((from, data) -> {
+ byte[] bytes = new byte[data.remaining()];
+ data.get(bytes);
+ String msg = new String(bytes);
+ System.out.println("Received from " + from + ": " + msg);
+ });
+
+ System.out.println("Echo server running on port " + port);
+ System.out.println("Press Enter to stop...");
+ System.in.read();
+
+ channel.close();
+ System.out.println("Server stopped.");
+ }
+}
\ No newline at end of file
diff --git a/src/org/kne/cloud/network/klalb/KLALBController.java b/src/org/kne/cloud/network/klalb/KLALBController.java
index 745238e..7e5d2e1 100644
--- a/src/org/kne/cloud/network/klalb/KLALBController.java
+++ b/src/org/kne/cloud/network/klalb/KLALBController.java
@@ -686,7 +686,7 @@ public class KLALBController {
context.syncToSystem();
// System.out.println(context);
try {
- NTPv4Protocol nvc = new NTPv4Protocol(context,
+ /*NTPv4Protocol nvc = new NTPv4Protocol(context,
new MultiProtocolSocketAddress(iproxyname + "datagram","0.0.0.0",123));
srv6Router.addSRv6RouterListener(new SRv6RouterListener() {
@@ -718,7 +718,7 @@ public class KLALBController {
return true;
});
}
- });
+ });*/
nvc2 = new NTPv4Protocol(context, new MultiProtocolSocketAddress("udp","0.0.0.0",0));
System.out.println(" Loaded: NTP Module");