netty_use

使用netty来简单写个demo,熟悉语法和常见的命令

1
2
3
4
5
6
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.49.Final</version>
</dependency>

client:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
public class ITDragonAIOClient implements Runnable{  
private static Integer PORT = 8888;
private static String IP_ADDRESS = "127.0.0.1";
private AsynchronousSocketChannel asynSocketChannel ;
public ITDragonAIOClient() throws Exception {
asynSocketChannel = AsynchronousSocketChannel.open(); // 打开通道
}
public void connect(){
asynSocketChannel.connect(new InetSocketAddress(IP_ADDRESS, PORT)); // 创建连接 和NIO一样
}
public void write(String request){
try {
asynSocketChannel.write(ByteBuffer.wrap(request.getBytes())).get();
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
asynSocketChannel.read(byteBuffer).get();
byteBuffer.flip();
byte[] respByte = new byte[byteBuffer.remaining()];
byteBuffer.get(respByte); // 将缓冲区的数据放入到 byte数组中
System.out.println(new String(respByte,"utf-8").trim());
} catch (Exception e) {
e.printStackTrace();
}
}
@Override
public void run() {
while(true){
}
}
public static void main(String[] args) throws Exception {
for (int i = 0; i < 10; i++) {
ITDragonAIOClient myClient = new ITDragonAIOClient();
myClient.connect();
new Thread(myClient, "myClient").start();
String []operators = {"+","-","*","/"};
Random random = new Random(System.currentTimeMillis());
String expression = random.nextInt(10)+operators[random.nextInt(4)]+(random.nextInt(10)+1);
myClient.write(expression);
}
}
}

server:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
/**
* AIO, 也叫 NIO2.0 是一种异步非阻塞的通信方式
* AIO 引入了异步通道的概念 AsynchronousServerSocketChannel和AsynchronousSocketChannel 其read和write方法返回值类型是Future对象。
*/
public class ITDragonAIOServer {

private ExecutorService executorService; // 线程池
private AsynchronousChannelGroup threadGroup; // 通道组
public AsynchronousServerSocketChannel asynServerSocketChannel; // 服务器通道
public void start(Integer port){
try {
// 1.创建一个缓存池
executorService = Executors.newCachedThreadPool();
// 2.创建通道组
threadGroup = AsynchronousChannelGroup.withCachedThreadPool(executorService, 1);
// 3.创建服务器通道
asynServerSocketChannel = AsynchronousServerSocketChannel.open(threadGroup);
// 4.进行绑定
asynServerSocketChannel.bind(new InetSocketAddress(port));
System.out.println("server start , port : " + port);
// 5.等待客户端请求
asynServerSocketChannel.accept(this, new ITDragonAIOServerHandler());
// 一直阻塞 不让服务器停止,真实环境是在tomcat下运行,所以不需要这行代码
Thread.sleep(Integer.MAX_VALUE);
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
ITDragonAIOServer server = new ITDragonAIOServer();
server.start(8888);
}
}

serverhandler:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
public class ITDragonAIOServerHandler implements CompletionHandler<AsynchronousSocketChannel, ITDragonAIOServer> {  
private final Integer BUFFER_SIZE = 1024;
@Override
public void completed(AsynchronousSocketChannel asynSocketChannel, ITDragonAIOServer attachment) {
// 保证多个客户端都可以阻塞
attachment.asynServerSocketChannel.accept(attachment, this);
read(asynSocketChannel);
}
//读取数据
private void read(final AsynchronousSocketChannel asynSocketChannel) {
ByteBuffer byteBuffer = ByteBuffer.allocate(BUFFER_SIZE);
asynSocketChannel.read(byteBuffer, byteBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer resultSize, ByteBuffer attachment) {
//进行读取之后,重置标识位
attachment.flip();
//获取读取的数据
String resultData = new String(attachment.array()).trim();
System.out.println("Server -> " + "收到客户端的数据信息为:" + resultData);
// String response = resultData + " = " + CalculatorUtil.cal(resultData);
String response = resultData + " = " ;
write(asynSocketChannel, response);
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
exc.printStackTrace();
}
});
}
// 写入数据
private void write(AsynchronousSocketChannel asynSocketChannel, String response) {
try {
// 把数据写入到缓冲区中
ByteBuffer buf = ByteBuffer.allocate(BUFFER_SIZE);
buf.put(response.getBytes());
buf.flip();
// 在从缓冲区写入到通道中
asynSocketChannel.write(buf).get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, ITDragonAIOServer attachment) {
exc.printStackTrace();
}
}