您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

yanf4j 0.50-beta2 版本釋出：引入客戶端非阻塞連接 API

編輯：Java 相關知識

yanf4j 框架最新發布了 0.50-beta2 版本，此次更新的核心亮點在於引入了客戶端非阻塞連接 API，專為高併發網絡編程場景設計。新增的兩大核心控制器 TCPConnectorController 和 UDPConnectorController 分別處理 TCP 和 UDP 協議的客戶端連接控制，極大地簡化了非阻塞網絡通訊的開發難度。

UDP 非阻塞客戶端實現

UDP 協議本質上是無連接的，因此 connect() 方法僅調用底層 DatagramChannel.connect() 來限制收發數據包的遠程端點。以下是一個簡潔的 UDP Echo Client 實現示例：

// 客戶端 Echo 處理器
class EchoClientHandler extends HandlerAdapter {
    public void onReceive(Session udpSession, Object t) {
        DatagramPacket datagramPacket = (DatagramPacket) t;
        System.out.println("recv:" + new String(datagramPacket.getData()));
    }
    
    @Override
    public void onMessageSent(Session session, Object t) {
        System.out.println("send:" + new String((byte[]) t));
    }
}

// 連接代碼並發送 UDP 包
UDPConnectorController connector = new UDPConnectorController();
connector.setSoTimeout(1000);
connector.setHandler(new EchoClientHandler());
connector.connect(new InetSocketAddress(InetAddress.getByName(host), port));

for (int i = 0; i < 10000; i++) {
    String s = "hello " + i;
    DatagramPacket packet = new DatagramPacket(s.getBytes(), s.length());
    connector.send(packet);
}

TCP 非阻塞客戶端實現

TCPConnectorController 的使用方式類似，但提供了更豐富的連接管理功能。關鍵在於 connect() 方法為非阻塞調用，立即返回，連接狀態可通過 isConnected() 檢查，最佳實踐是在 Handler 的 onConnected() 回調中處理後續邏輯：

// 客戶端 Echo 處理器
class EchoHandler extends HandlerAdapter<String> {
    @Override
    public void onConnected(Session session) {
        try {
            for (int i = 0; i < NUM; i++) {
                session.send(generateString(i));
            }
        } catch (Exception e) {
            // 異常處理
        }
    }
    
    public String generateString(int len) {
        StringBuffer sb = new StringBuffer();
        for (int i = 0; i < MESSAGE_LEN; i++) sb.append(i);
        return sb.toString();
    }
    
    @Override
    public void onReceive(Session session, String t) {
        if (DEBUG) System.out.println("recv:" + t);
    }
}

// TCP 連接示例
Configuration configuration = new Configuration();
configuration.setTcpSessionReadBufferSize(256 * 1024);
TCPConnectorController connector = new TCPConnectorController(configuration, new StringCodecFactory());
connector.setHandler(new EchoHandler());
connector.setCodecFactory(new StringCodecFactory());

try {
    connector.connect(new InetSocketAddress("localhost", 8080));
    // 推薦在 onConnected 回調中處理邏輯，而非輪詢 isConnected()
} catch (IOException e) {
    e.printStackTrace();
}

核心發送 API

兩個 ConnectorController 均提供系列發送方法：

TCPConnectorController.send(Object msg) throws InterruptedException
UDPConnectorController.send(DatagramPacket packet) throws InterruptedException
UDPConnectorController.send(SocketAddress targetAddr, Object msg) throws InterruptedException

重要新特性：會話發送隊列流量控制

0.50-beta 版本引入了會話發送隊列緩衝區的流量控制選項，默認情況下發送隊列是無界且非阻塞的。啟用高低水位標記後，流量控制規則如下：

當發送隊列數據總量超過高水位標記（highWaterMark），Session.send 將阻塞
阻塞持續到隊列數據總量低於低水位標記（lowWaterMark）才解除

相關配置方法：

public void setSessionWriteQueueHighWaterMark(int highWaterMark);
public void setSessionWriteQueueLowWaterMark(int lowWaterMark);

此設計靈感來源於 ACE 的 ACE_Message_Queue，由 com.google.code.yanf4j.util.MessageQueue 實現。

其他增強功能

Session 接口新增 setReadBufferByteOrder 方法：用於設置接收緩衝區字節序，默認為網絡字節序（大端模式），建議在 onSessionStarted 回調中調用
Session.send(Object msg, long timeout) 重載方法：支持超時控制，超時後終止發送
發送方法返回布爾值：表示發送成功與否，並在啟用流量控制時拋出 InterruptedException 響應中斷

這些更新使 yanf4j 在高性能網絡應用開發中更加得心應手，特別適合需要精細控制網絡流量和連接生命週期的場景。