Java網關服務-AIO(三)

概述

前兩節中，我們已經獲取了body的總長度，剩下的就是讀出body，處理請求

ChannelServerHandler

ChannelServerHandler即從channel中讀取請求，也向channle輸出結果，因此它實現了InboundHandler, OutboundHandler

/**

* 讀取請求的內容，業務處理

*/

public class ChannelServerHandler implements CompletionHandler, InboundHandler, OutboundHandler {

private final static Logger LOGGER = LoggerFactory.getLogger(ChannelServerHandler.class);

private AsynchronousSocketChannel channel;

public ChannelServerHandler(AsynchronousSocketChannel channel) {

this.channel = channel;

}

public void completed(Integer result, ByteBuffer attachment) {

//如果條件成立，說明客戶端主動終止了TCP套接字，這時服務端終止就可以了

if (result == -1) {

System.out.println("remote is close");

closeChannel();

return;

}

Object resultData;

String req = (String) read(channel, attachment);

if (req == null) {

closeChannel();

return;

}

try {

LOGGER.info("socket:{}", channel.getRemoteAddress());

//同步處理請求

RequestHandler requestHandler = ApplicationUtils.getBean(RequestHandler.class);

resultData = requestHandler.execute(req);

} catch (Throwable t) {

resultData = Result.error("ERROR", Utils.error(t));

LOGGER.error("調用接口失敗", t);

}

if (resultData == null) {

resultData = Result.failure("FAILURE", "調用失敗,數據為空.");

}

try {

String resultContent = resultData instanceOf String ? (String) resultData : JSON.toJSONString(resultData);

byte[] bytes = resultContent.getBytes("UTF-8");

ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);

writeBuffer.put(bytes);

writeBuffer.flip();

write(channel, writeBuffer);

} catch (Exception e) {

LOGGER.error("對象轉JSON失敗,對象:{}", resultData, e);

}

closeChannel();

}

@Override

public Object read(AsynchronousSocketChannel socketChannel, ByteBuffer in) {

in.flip();

byte[] body = new byte[in.remaining()];

in.get(body);

String req = null;

try {

req = new String(body, "UTF-8");

} catch (UnsupportedEncodingException e) {

e.printStackTrace();

}

return req;

}

@Override

public Object write(AsynchronousSocketChannel socketChannel, ByteBuffer out) {

//write，write操作結束后關閉通道

channel.write(out, out, new CompletionHandler() {

@Override

public void completed(Integer result, ByteBuffer attachment) {

closeChannel();

}

@Override

public void failed(Throwable exc, ByteBuffer attachment) {

closeChannel();

}

});

return null;

}

public void failed(Throwable exc, ByteBuffer attachment) {

closeChannel();

}

private void closeChannel() {

try {

this.channel.close();

} catch (IOException e) {

e.printStackTrace();

}

}

}

讀取body

in.flip();

byte[] body = new byte[in.remaining()];

in.get(body);

String req = null;

try {

req = new String(body, "UTF-8");

} catch (UnsupportedEncodingException e) {

e.printStackTrace();

}

return req;

in.remaining()

buffer中含有的字節數

客戶端、服務端由于跨語言和經驗問題，沒有使用復雜的跨語言序列化技術，雙方約定使用UTF-8編碼，通過將body轉換為String，最終獲得了客戶端傳遞的字符串。

處理請求

經過自定義的請求處理邏輯，同步處理，最終將響應編碼后，發送給客戶端，write操作結束后，關閉連接

總結

使用AIO開發服務端時，主要涉及

配置I/O事件完成的回調線程池

從accept -> read 到 向client端響應 write -> close,盡量使用CompletionHanlder來異步處理，不要在處理某個事件完成的線程中，同步的調用，如future.get()

如果是短連接，則需在write操作時注冊write結束后的handler，在handler中關閉連接

擴展

長連接該如何處理

長連接意味著client可以發多次請求，由于多次請求被server執行的順序是不可控的，可能后發的請求先響應，因此需要在請求和響應時，加requestId，據此對應到請求的結果

長連接不需要在write后關閉連接

長連接需要開發定時的ping-pong心跳消息

長連接在響應時比現在更復雜，也需要一個和請求類似或相同的協議來標識body長度

測試

測試用例

/**

* mvn -Dtest=com.jd.jshop.web.sdk.test.ClientTest#pingReqSocket test

*

* @throws IOException

*/

@Test

@PerfTest(invocations = 20000, threads = 50)

public void pingReqSocket() throws IOException {

byte[] content = "ping".getBytes("UTF-8");

String result = sendReq(content);

//斷言 是否和預期一致

Assert.assertEquals("pong", result);

}

private String sendReq(byte[] content) throws IOException {

ByteBuffer writeBuffer = ByteBuffer.allocate(4 + content.length);

writeBuffer.putInt(content.length);

writeBuffer.put(content);

writeBuffer.flip();

Socket socket = new Socket();

socket.connect(new InetSocketAddress("127.0.0.1", 9801));

socket.getOutputStream().write(writeBuffer.array());

socket.getOutputStream().flush();

byte[] buf = new byte[1024];

int len = 0;

String result = null;

while ((len = socket.getInputStream().read(buf)) != -1) {

result = new String(buf, 0, len);

System.out.println(result);

}

return result;

}

測試的方法是，在服務器上建立socket連接，向server發送ping，server返回pong

測試服務器：centos, 2個物理核，4個邏輯核，內存16G

分析aio的實現：

在ping-pong測試中性能極高，優于并甩開netty

以下是使用Netty開發的server端的測試用例，可以和上面的圖片對比一下

Measured invocations:10,000

Thread Count:10

Measured

(system)Required

Execution time:1,646 ms

Throughput:6,075 / s

Min. latency:0 ms

Average latency:1 ms

Median:2 ms

90%:2 ms

Max latency:26 ms

============================

Started at:Oct 16, 2018 5:27:03 PM

Measured invocations:20,000

Thread Count:20

Measured

(system)Required

Execution time:3,293 ms

Throughput:6,073 / s

Min. latency:0 ms

Average latency:3 ms

Median:3 ms

90%:5 ms

Max latency:54 ms

============================

Started at:Oct 16, 2018 5:28:24 PM

Measured invocations:20,000

Thread Count:10

Measured

(system)Required

Execution time:3,051 ms

Throughput:6,555 / s

Min. latency:0 ms

Average latency:1 ms

Median:1 ms

90%:2 ms

Max latency:44 ms

============================

Started at:Oct 16, 2018 5:30:06 PM

Measured invocations:20,000

Thread Count:50

Measured

(system)Required

Execution time:3,167 ms

Throughput:6,315 / s

Min. latency:0 ms

Average latency:7 ms

Median:7 ms

90%:10 ms

Max latency:64 ms

分析基于Netty的實現：

吞吐量：6000+/s

10個線程時：90%低于2ms,平均1ms

20個線程時：90%低于5ms，平均3ms

50個線程時：90%低于10ms，平均7ms

線程越多，性能越低

當前測試用例不太依賴內存

執行10000+次請求，建立10000+連接，要求服務器對單個進程fd限制打開,防止報too many open files導致測試用例執行失敗

ulimit -n 20240

總結

以上是生活随笔為你收集整理的javaweb网关_Java网关服务-AIO（三）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。