Java NIO 基础
This commit is contained in:
罗祥
24
code/Java/nio-tutorial/pom.xml
Normal file
24
code/Java/nio-tutorial/pom.xml
Normal file
@ -0,0 +1,24 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<groupId>org.example</groupId>
|
||||
<artifactId>nio-tutorial</artifactId>
|
||||
<version>1.0-SNAPSHOT</version>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<configuration>
|
||||
<target>1.8</target>
|
||||
<source>1.8</source>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
133
code/Java/nio-tutorial/src/main/java/chat/ChatClient.java
Normal file
133
code/Java/nio-tutorial/src/main/java/chat/ChatClient.java
Normal file
@ -0,0 +1,133 @@
|
||||
package chat;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.ClosedSelectorException;
|
||||
import java.nio.channels.SelectionKey;
|
||||
import java.nio.channels.Selector;
|
||||
import java.nio.channels.SocketChannel;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Scanner;
|
||||
import java.util.Set;
|
||||
|
||||
public class ChatClient {
|
||||
|
||||
private String hostname;
|
||||
private int port;
|
||||
private Selector selector;
|
||||
private final ByteBuffer rBuffer = ByteBuffer.allocate(1024);
|
||||
private final ByteBuffer wBuffer = ByteBuffer.allocate(1024);
|
||||
|
||||
ChatClient(String hostname, int port) {
|
||||
this.hostname = hostname;
|
||||
this.port = port;
|
||||
}
|
||||
|
||||
public void start() {
|
||||
try {
|
||||
// 创建SocketChannel
|
||||
SocketChannel socketChannel = SocketChannel.open();
|
||||
socketChannel.configureBlocking(false);
|
||||
socketChannel.connect(new InetSocketAddress(hostname, port));
|
||||
|
||||
// 创建selector
|
||||
selector = Selector.open();
|
||||
// 注册监听CONNECT事件
|
||||
socketChannel.register(selector, SelectionKey.OP_CONNECT);
|
||||
|
||||
// 持续调用select
|
||||
while (true) {
|
||||
selector.select();
|
||||
Set<SelectionKey> selectionKeys = selector.selectedKeys();
|
||||
for (SelectionKey selectionKey : selectionKeys) {
|
||||
// 处理Connect事件
|
||||
if (selectionKey.isConnectable()) {
|
||||
SocketChannel clientChannel = (SocketChannel) selectionKey.channel();
|
||||
// 有可能正处于连接中的状态
|
||||
if (clientChannel.isConnectionPending()) {
|
||||
// 等待连接完成
|
||||
clientChannel.finishConnect();
|
||||
// 开始监听用户输入
|
||||
inputListening(clientChannel);
|
||||
// 为clientChannel注册上Read
|
||||
clientChannel.register(selector, SelectionKey.OP_READ);
|
||||
System.out.println("尊敬的用户" + clientChannel.socket().getLocalPort() + ", 你已成功加入群聊！");
|
||||
}
|
||||
// 处理Read事件
|
||||
} else if (selectionKey.isReadable()) {
|
||||
handleMessage(selectionKey);
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (ClosedSelectorException e) {
|
||||
// 主动关闭客户端，不做任何处理
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
} finally {
|
||||
try {
|
||||
selector.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 监听用户来自控制台的输入
|
||||
*
|
||||
* @param clientChannel 客户端Channel
|
||||
*/
|
||||
private void inputListening(SocketChannel clientChannel) {
|
||||
// 要持续监听用户输入，但又不能阻塞主线程，所以需要一个单独的线程来完成
|
||||
new Thread(() -> {
|
||||
Scanner scanner = new Scanner(System.in);
|
||||
try {
|
||||
while (scanner.hasNextLine()) {
|
||||
String input = scanner.nextLine();
|
||||
if (!input.isEmpty()) {
|
||||
if ("exit".equals(input)) {
|
||||
selector.close();
|
||||
return;
|
||||
}
|
||||
wBuffer.put(input.getBytes());
|
||||
wBuffer.flip();
|
||||
while (wBuffer.hasRemaining()) {
|
||||
clientChannel.write(wBuffer);
|
||||
}
|
||||
wBuffer.clear();
|
||||
}
|
||||
}
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}).start();
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理来自其他客户端的消息
|
||||
*
|
||||
* @param selectionKey 处于Read状态的selectionKey
|
||||
*/
|
||||
private void handleMessage(SelectionKey selectionKey) {
|
||||
try {
|
||||
SocketChannel channel = (SocketChannel) selectionKey.channel();
|
||||
StringBuilder buffer = new StringBuilder();
|
||||
while (channel.read(rBuffer) > 0) {
|
||||
rBuffer.flip();
|
||||
buffer.append(StandardCharsets.UTF_8.decode(rBuffer));
|
||||
rBuffer.clear();
|
||||
}
|
||||
System.out.println(buffer.toString());
|
||||
} catch (IOException e) {
|
||||
selectionKey.cancel();
|
||||
selector.wakeup();
|
||||
System.out.println("聊天室服务器已关闭！");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static void main(String[] args) {
|
||||
new ChatClient("127.0.0.1", 8888).start();
|
||||
}
|
||||
}
|
||||
134
code/Java/nio-tutorial/src/main/java/chat/ChatServer.java
Normal file
134
code/Java/nio-tutorial/src/main/java/chat/ChatServer.java
Normal file
@ -0,0 +1,134 @@
|
||||
package chat;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.*;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Set;
|
||||
|
||||
public class ChatServer {
|
||||
|
||||
private String hostname;
|
||||
private int port;
|
||||
private Selector selector;
|
||||
private final ByteBuffer rBuffer = ByteBuffer.allocate(1024);
|
||||
private final ByteBuffer wBuffer = ByteBuffer.allocate(1024);
|
||||
|
||||
ChatServer(int port) {
|
||||
this("127.0.0.1", port);
|
||||
}
|
||||
|
||||
ChatServer(String hostname, int port) {
|
||||
this.hostname = hostname;
|
||||
this.port = port;
|
||||
}
|
||||
|
||||
public void start() {
|
||||
try {
|
||||
// 创建ServerSocketChannel
|
||||
ServerSocketChannel serverChannel = ServerSocketChannel.open();
|
||||
// 设置为非阻塞模式
|
||||
serverChannel.configureBlocking(false);
|
||||
serverChannel.bind(new InetSocketAddress(hostname, port));
|
||||
|
||||
// 创建selector
|
||||
selector = Selector.open();
|
||||
// 注册监听CONNECT事件
|
||||
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
|
||||
|
||||
// 持续调用select()
|
||||
while (true) {
|
||||
selector.select();
|
||||
Set<SelectionKey> selectionKeys = selector.selectedKeys();
|
||||
for (SelectionKey selectionKey : selectionKeys) {
|
||||
// 处理Accept事件
|
||||
if (selectionKey.isAcceptable()) {
|
||||
// 获取ServerSocketChannel
|
||||
ServerSocketChannel server = (ServerSocketChannel) selectionKey.channel();
|
||||
// 通过ServerSocketChannel获取SocketChannel
|
||||
SocketChannel clientChannel = server.accept();
|
||||
clientChannel.configureBlocking(false);
|
||||
// 注册Read事件
|
||||
clientChannel.register(selector, SelectionKey.OP_READ);
|
||||
System.out.println("欢迎" + clientChannel.socket().getPort() + "加入聊天室！");
|
||||
// 处理Readable事件
|
||||
} else if (selectionKey.isReadable()) {
|
||||
handleMessage(selectionKey);
|
||||
}
|
||||
}
|
||||
selectionKeys.clear();
|
||||
}
|
||||
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
} finally {
|
||||
// 关闭selector后，与之相关的所有资源都会被释放
|
||||
try {
|
||||
selector.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理客户端消息
|
||||
*
|
||||
* @param selectionKey 处于Read状态的SelectionKey
|
||||
*/
|
||||
private void handleMessage(SelectionKey selectionKey) {
|
||||
SocketChannel clientChannel = (SocketChannel) selectionKey.channel();
|
||||
try {
|
||||
// 获取来自客户端的消息
|
||||
StringBuilder buffer = new StringBuilder();
|
||||
while (clientChannel.read(rBuffer) > 0) {
|
||||
rBuffer.flip();
|
||||
buffer.append(StandardCharsets.UTF_8.decode(rBuffer));
|
||||
rBuffer.clear();
|
||||
}
|
||||
String msg = buffer.toString();
|
||||
// 将消息发送给其他客户端
|
||||
broadcastMessage(clientChannel, msg);
|
||||
} catch (IOException e) {
|
||||
// 该异常由clientChannel.read(rBuffer)方法抛出，如果出现该异常，则说明clientChannel已经关闭
|
||||
// 此时需要调用cancel()取消注册在selector上的事件
|
||||
selectionKey.cancel();
|
||||
// 在多线程环境下，如果另一个线程正在阻塞地调用select()，因为事件集已经改变，因此需要通过wakeup()让其立刻返回并重新select()
|
||||
selector.wakeup();
|
||||
System.out.println("用户" + clientChannel.socket().getPort() + "退出聊天室！");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将客户端发来的消息广播给其他客户端
|
||||
*
|
||||
* @param clientChannel 消息源
|
||||
* @param msg 消息
|
||||
*/
|
||||
private void broadcastMessage(SocketChannel clientChannel, String msg) {
|
||||
try {
|
||||
// 遍历当前selector上所有channel
|
||||
for (SelectionKey selectionKey : selector.keys()) {
|
||||
SelectableChannel channel = selectionKey.channel();
|
||||
// 消息不需要转发给ServerSocketChannel和当前客户端自己
|
||||
if (selectionKey.isValid() && !(channel instanceof ServerSocketChannel) && !channel.equals(clientChannel)) {
|
||||
SocketChannel otherClient = (SocketChannel) channel;
|
||||
wBuffer.put(("用户" + clientChannel.socket().getPort() + ": " + msg).getBytes());
|
||||
while (wBuffer.hasRemaining()) {
|
||||
wBuffer.flip();
|
||||
otherClient.write(wBuffer);
|
||||
}
|
||||
wBuffer.clear();
|
||||
}
|
||||
}
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
new ChatServer(8888).start();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user