简单版本AIO聊天室
实现一个简单的回声室服务器：服务器会简单地将客户端发送的消息原封不动地发送给客户端。
Server实现（CompletionHandler实现AIO）
import java.io.*;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.HashMap;
import java.util.Map;public class Server {final String LOCALHOST = "localhost";final int DEFAULT_PORT = 8888;AsynchronousServerSocketChannel serverChannel;private void close(Closeable closable) {if (closable != null) {try {closable.close();System.out.println("关闭" + closable);} catch (IOException e) {e.printStackTrace();}}}public void start() {try {// 绑定监听端口// 使用默认的AsynchronousChannelGroupserverChannel = AsynchronousServerSocketChannel.open();serverChannel.bind(new InetSocketAddress(LOCALHOST, DEFAULT_PORT));System.out.println("启动服务器，监听端口:" + DEFAULT_PORT);while (true) {// 当serverChannel连接到客户端的连接后，会调用AcceptHandler()来处理// 这一操作是不阻塞的， 只有连接完成后才会调用AcceptHandler()来处理serverChannel.accept(null, new AcceptHandler());// System.in.read()是阻塞的，防止提交过多serverChannel.accept的任务System.in.read();}} catch (IOException e) {e.printStackTrace();} finally {close(serverChannel);}}private class AcceptHandler implementsCompletionHandler<AsynchronousSocketChannel, Object>{@Override// AsynchronousSocketChannel result中的result是serverChannel.accept连接成功后返回的结果public void completed(AsynchronousSocketChannel result, Object attachment) {if (serverChannel.isOpen()) {// 异步等待其它客户端的连接serverChannel.accept(null, this);}AsynchronousSocketChannel clientChannel = result;if (clientChannel != null && clientChannel.isOpen()) {ClientHandler handler = new ClientHandler(clientChannel);ByteBuffer buffer = ByteBuffer.allocate(1024);Map<String, Object> info = new HashMap<>();info.put("type", "read");info.put("buffer", buffer);// 向buffer中写入数据，buffer为写状态clientChannel.read(buffer, info, handler);}}@Overridepublic void failed(Throwable exc, Object attachment) {// 处理错误}}private class ClientHandler implementsCompletionHandler<Integer, Object>{private AsynchronousSocketChannel clientChannel;public ClientHandler(AsynchronousSocketChannel channel) {this.clientChannel = channel;}@Overridepublic void completed(Integer result, Object attachment) {Map<String, Object> info = (Map<String, Object>) attachment;String type = (String) info.get("type");if ("read".equals(type)) {ByteBuffer buffer = (ByteBuffer) info.get("buffer");// 即将向buffer读取数据，所以要将buffer的状态改变为读状态buffer.flip();info.put("type", "write");clientChannel.write(buffer, info, this);buffer.clear();} else if ("write".equals(type)) {ByteBuffer buffer = ByteBuffer.allocate(1024);info.put("type", "read");info.put("buffer", buffer);clientChannel.read(buffer, info, this);}}@Overridepublic void failed(Throwable exc, Object attachment) {// 处理错误}}public static void main(String[] args) {Server server = new Server();server.start();}
}Client实现（Future实现AIO）
import java.io.*;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;public class Client {final String LOCALHOST = "localhost";final int DEFAULT_PORT = 8888;AsynchronousSocketChannel clientChannel;private void close(Closeable closable) {if (closable != null) {try {closable.close();System.out.println("关闭" + closable);} catch (IOException e) {e.printStackTrace();}}}public void start() {try {// 创建channelclientChannel = AsynchronousSocketChannel.open();Future<Void> future = clientChannel.connect(new InetSocketAddress(LOCALHOST, DEFAULT_PORT));future.get();// 等待用户的输入BufferedReader consoleReader =new BufferedReader(new InputStreamReader(System.in));while (true) {String input = consoleReader.readLine();byte[] inputBytes = input.getBytes();ByteBuffer buffer = ByteBuffer.wrap(inputBytes);Future<Integer> writeResult = clientChannel.write(buffer);writeResult.get();buffer.flip();Future<Integer> readResult = clientChannel.read(buffer);readResult.get();String echo = new String(buffer.array());buffer.clear();System.out.println(echo);}} catch (IOException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();} finally {close(clientChannel);}}public static void main(String[] args) {Client client = new Client();client.start();}}要点：
- 在Java NIO中，使用AsynchronousSocketChannel的read方法时，即使clientChannel中没有立即可用的数据，read操作也会被异步提交，并且不会阻塞。这意味着read操作会被注册到系统中，然后方法会立即返回，而不会等待数据的到来。（write操作也是如此）
- 通道Channel中的数据会暂时存储在通道的内部缓冲区中，直到被读取后才会从Buffer中移除
- 当客户端发来一段消息时，服务器端只会调用两次ClientHandler中的completed方法。因为当Channel中的数据被读完后，没有数据就无法完成clientChannel.read的操作，自然不会再次调用ClientHandler中的completed方法
- 每个ClientHandler实例都是为一个特定的客户端连接创建的，并且只被用来处理该连接的读写操作。这意味着对于每个客户端连接，只有一个ClientHandler实例，并且只有一个线程（NIO线程）会调用该实例的completed方法。因此，在这种情况下，info的修改是线程安全的，因为对于每个ClientHandler实例，没有多个线程同时对其进行修改。（可以理解为每个Handler中创建一个为特定客户端的map）
AIO模型多人聊天室
在简单版本AIO聊天室的基础上，为服务器端增加了消息转发功能
Server实现
import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class ChatServer {private static final String LOCALHOST = "localhost";private static final int DEFAULT_PORT = 8888;private static final String QUIT = "quit";private static final int BUFFER_SIZE = 1024;private static final int THREADPOOL_SIZE = 8;private AsynchronousChannelGroup channelGroup;private AsynchronousServerSocketChannel serverSocketChannel;private List<ClientHandler> connectedClients;private Charset charset = Charset.forName("UTF-8");private int port;// 注意初始化List，防止空指针报错public ChatServer() {this(DEFAULT_PORT);connectedClients = new ArrayList<>();}public ChatServer(int port) {connectedClients = new ArrayList<>();this.port = port;}private Boolean readyToQuit(String msg) {return QUIT.equals(msg);}private void close(Closeable closeable) {if (closeable != null) {try {closeable.close();} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}private void start() {ExecutorService executorService = Executors.newFixedThreadPool(THREADPOOL_SIZE);try {channelGroup = AsynchronousChannelGroup.withThreadPool(executorService);serverSocketChannel = AsynchronousServerSocketChannel.open(channelGroup);serverSocketChannel.bind(new InetSocketAddress(LOCALHOST, port));System.out.println("启动服务器，监听端口：" + port);while (true) {serverSocketChannel.accept(null, new AcceptHandler());System.in.read();}} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();}}private class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> {@Overridepublic void completed(AsynchronousSocketChannel clientChannel, Object attachment) {if (serverSocketChannel.isOpen()) {serverSocketChannel.accept(null, this);}if (clientChannel != null && clientChannel.isOpen()) {ClientHandler handler = new ClientHandler(clientChannel);ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);addClient(handler);clientChannel.read(buffer, buffer, handler);}}@Overridepublic void failed(Throwable exc, Object attachment) {System.out.println("connect faild");}}private synchronized void addClient(ClientHandler handler) {connectedClients.add(handler);System.out.println(getClientName(handler.clientChannel) + "已连接");}private synchronized void removeClient(ClientHandler handler) {connectedClients.remove(handler);System.out.println(getClientName(handler.clientChannel) + "已断开连接");}private class ClientHandler implements CompletionHandler<Integer, Object> {private AsynchronousSocketChannel clientChannel;public ClientHandler(AsynchronousSocketChannel clientChannel) {this.clientChannel = clientChannel;}@Overridepublic void completed(Integer result, Object attachment) {ByteBuffer buffer = (ByteBuffer) attachment;if (buffer != null) {if (result <= 0) {// 客户端异常// 将客户移除出在线客户列表removeClient(this);} else {buffer.flip();String fwdMsg = receive(buffer);System.out.println(getClientName(clientChannel) + fwdMsg);forwardMessage(clientChannel, fwdMsg);buffer.clear();if (readyToQuit(fwdMsg)) {removeClient(this);} else {clientChannel.read(buffer, buffer, this);}}}}@Overridepublic void failed(Throwable exc, Object attachment) {System.out.println("error");}}private synchronized void forwardMessage(AsynchronousSocketChannel clientChannel, String msg) {for (ClientHandler handler : connectedClients) {if (!clientChannel.equals(handler.clientChannel)) {ByteBuffer buffer = charset.encode(getClientName(handler.clientChannel) + ":" + msg);handler.clientChannel.write(buffer, null, handler);} }}private String getClientName(AsynchronousSocketChannel clientChannel) {int clientPort = -1;try {InetSocketAddress address = (InetSocketAddress) clientChannel.getRemoteAddress();clientPort = address.getPort();} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();}return "客户端【" + clientPort + "】";}private String receive(ByteBuffer buffer) {CharBuffer charBuffer = charset.decode(buffer);return String.valueOf(charBuffer);}public static void main(String[] args) {ChatServer server = new ChatServer(DEFAULT_PORT);server.start();}
}
Client实现
ChatClient:
import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;public class ChatClient {private static final String LOCALHOST = "localhost";private static final int DEFAULT_PORT = 8888;private static final String QUIT = "quit";private static final int BUFFER_SIZE = 1024;private String host;private int port;private AsynchronousSocketChannel clientChannel;private Charset charset = Charset.forName("UTF-8");public ChatClient() {this(LOCALHOST, DEFAULT_PORT);}public ChatClient(String host, int port) {this.host = host;this.port = port;}public boolean readyToQuit(String msg) {return QUIT.equals(msg);}private void close(Closeable closeable) {if (closeable != null) {try {closeable.close();} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}private void start() {// 创建Channeltry {clientChannel = AsynchronousSocketChannel.open();System.out.println(port);Future<Void> future = clientChannel.connect(new InetSocketAddress(host, DEFAULT_PORT));future.get();// 处理用户的输入new Thread(new UserInputHandler(this)).start();ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);while (true) {Future<Integer> readResult = clientChannel.read(buffer);int result = readResult.get();if (result <= 0) {// 服务器异常System.out.println("服务器断开");close(clientChannel);System.exit(1);} else {buffer.flip();String msg = String .valueOf(charset.decode(buffer));buffer.clear();System.out.println(msg);}}} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (ExecutionException e) {// TODO Auto-generated catch blocke.printStackTrace();}}public void send(String msg) {if (msg.isEmpty()) {return ;}ByteBuffer buffer = charset.encode(msg);Future<Integer> future = clientChannel.write(buffer);try {future.get();} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (ExecutionException e) {// TODO Auto-generated catch blocke.printStackTrace();}}public static void main(String[] args) {ChatClient client = new ChatClient(LOCALHOST, DEFAULT_PORT);client.start();}
}UserInputHandler:
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;public class UserInputHandler implements Runnable{private ChatClient chatClient;public UserInputHandler(ChatClient chatClient) {this.chatClient = chatClient;}@Overridepublic void run() {BufferedReader consoleReader = new BufferedReader(new InputStreamReader(System.in));while (true) {String input;try {input = consoleReader.readLine();chatClient.send(input);if (chatClient.readyToQuit(input)) {break;}} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}}
