Netty条记3：NIO编程 - IT评测·应用市场-qidao123.com技术社区

public class ServerHandle implements Runnable{
private Selector selector;
private ServerSocketChannel serverSocketChannel;
public ServerHandle(int port) {
try {
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
// channel必须处于非阻塞模式下，不然会报错，所以不能同FileChannel一起使用
serverSocketChannel.configureBlocking(false);
serverSocketChannel.socket().bind(new InetSocketAddress(port));
// 注册对应的事件，这里注册的是accept事件，只要监听到就会调用对应的处理器
// 注册还可以添加附加对象，也就是第三个参数，获取方式：key.attachment();
// 如果注册了事件，需要取消，需要调用channel.cancel()
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("服务端已准备好：" + port);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public void run() {
while (true) {
try {
// 阻塞直到通道就绪，这边设置了超时时间
// 返回值：有多少通道就绪
selector.select(1000);
// 在通道就绪时，获取对应的键，也就是事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
// 根据事件的类型进行对应的处理
handlerInput(key);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
private void handlerInput(SelectionKey key) throws IOException {
// 在这个循环中，可能存在key.cancel() 或者移除，所以这里需要判断是否有效
if (!key.isValid()) {
return;
}
try {
if (key.isAcceptable()) {
// 这里处理客户端连接服务端的事件
ServerSocketChannel channel = (ServerSocketChannel)key.channel();
try {
// 接收请求
SocketChannel sc = channel.accept();
System.out.println("-----建立连接------");
// 设置该通道非阻塞
sc.configureBlocking(false);
// 并注册read事件，监听着
sc.register(selector, SelectionKey.OP_READ);
} catch (IOException e) {
System.out.println("连接客户端失败！");
key.cancel();
channel.close();
}
}
if (key.isReadable()) {
SocketChannel sc = (SocketChannel)key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
int read = sc.read(buffer);
if (read > 0) {
// 反转：将这个缓冲中的数据从现在的位置变成从0开始
buffer.flip();
byte[] bytes = new byte[buffer.remaining()];
buffer.get(bytes);
String msg = new String(bytes, StandardCharsets.UTF_8);
System.out.println("服务器收到消息：" + msg);
doWrite(sc, "hello，收到消息：" + msg);
}
}
} catch (IOException e) {
System.out.println("数据处理失败！");
// 再处理失败，或异常时要退出通道，不然每次循环都会检查通道导致一致报错
key.channel().close();
key.cancel();
}
}
private void doWrite(SocketChannel sc, String msg) throws IOException {
byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
buffer.put(bytes);
buffer.flip();
sc.write(buffer);
}
}

复制代码

public static void main(String[] args) {
ServerHandle serverHandle = new ServerHandle(8080);
new Thread(serverHandle).start();
}

复制代码

public class ClientHandle implements Runnable{
private String ip;
private int port;
private SocketChannel socketChannel;
private Selector selector;
public ClientHandle(String ip, int port) {
try {
this.ip = ip;
this.port = port;
selector = Selector.open();
socketChannel = SocketChannel.open();
// 非阻塞状态
socketChannel.configureBlocking(false);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public void run() {
try {
// 启动后，执行连接操作
if (socketChannel.connect(new InetSocketAddress(ip, port))) {
// 连接服务端成功后，注册读取事件
socketChannel.register(selector, SelectionKey.OP_READ);
} else {
// 如果连接失败，则再注册连接事件，之后再进行处理
socketChannel.register(selector, SelectionKey.OP_CONNECT);
}
} catch (IOException e) {
throw new RuntimeException(e);
}
while (true) {
try {
// 阻塞1000秒直到有通道就绪
selector.select(1000);
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
handleInput(key);
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
public void sendMsg(String msg) throws IOException {
byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
buffer.put(bytes);
buffer.flip();
socketChannel.write(buffer);
}
private void handleInput(SelectionKey key) throws IOException {
if (!key.isValid()) {
return;
}
SocketChannel sc = (SocketChannel) key.channel();
if (key.isConnectable()) {
if (sc.finishConnect()) {
socketChannel.register(selector, SelectionKey.OP_READ);
} else {
System.exit(1);
}
}
if (key.isReadable()) {
ByteBuffer buffer = ByteBuffer.allocate(1024);
int read = sc.read(buffer);
if (read > 0) {
buffer.flip();
byte[] bytes = new byte[buffer.remaining()];
buffer.get(bytes);
String msg = new String(bytes, StandardCharsets.UTF_8);
System.out.println("客户端收到消息：" + msg);
} else if (read < 0) {
key.cancel();;
sc.close();
}
}
}
}

复制代码

public static void main(String[] args) throws IOException {
ClientHandle handle = new ClientHandle("localhost", 8080);
new Thread(handle).start();
Scanner scanner = new Scanner(System.in);
// 死循环保持监听
while (true) {
// 每次控制台输入，就发送给服务端
handle.sendMsg(scanner.nextLine());
}
}

复制代码