NIO 实现非阻塞 Socket 通讯 - IT评测·应用市场-qidao123.com

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
/**
* 聊天室服务端
*/
public class NServer {
private Selector selector = null;
static final int PORT = 30000;
private Charset charset = Charset.forName("UTF-8");
ServerSocketChannel server = null;
public void init() throws IOException {
selector = Selector.open();
server = ServerSocketChannel.open();
InetSocketAddress isa = new InetSocketAddress("127.0.0.1", PORT);
// 将该 ServerSocketChannel 绑定到指定 IP 地址
server.bind(isa);
//设置为以非阻塞方式工作
server.configureBlocking(false);
// 将 server 注册到指定的 Selector
server.register(selector, SelectionKey.OP_ACCEPT);
while (selector.select() > 0) {
for (SelectionKey sk : selector.selectedKeys()) {
// 从 selector 上的已选择 Key 集中删除正在处理的 SelectionKey
selector.selectedKeys().remove(sk);
// 如果 sk 对应 channel 包含客户端的连接请求
if (sk.isAcceptable()) {
// 接受请求
SocketChannel sc = server.accept();
// 采用非阻塞模式
sc.configureBlocking(false);
// 将该 SocketChannel 也注册到 selector
sc.register(selector, SelectionKey.OP_READ);
// 将 sk 对应的 channel 设置成准备接受其他请求
sk.interestOps(SelectionKey.OP_ACCEPT);
}
// 如果 sk 对应的channel 有数据需要读取
if (sk.isReadable()) {
SocketChannel sc = (SocketChannel) sk.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
String content = "";
try {
// 读取数据操作
while (sc.read(buffer) > 0) {
buffer.flip();
content += charset.decode(buffer);
}
System.out.println("读取的数据： " + content);
sk.interestOps(SelectionKey.OP_READ);
} catch (IOException e) {
sk.cancel();
if (sk.channel() != null) {
sk.channel().close();
}
}
// 如果 content 的长度大于 0，即聊天信息不为空
if (content.length() > 0) {
// 遍历该 selector 里注册的所有 SelectionKey
for (SelectionKey key : selector.keys()) {
// 获取 channel
Channel targetChannel = key.channel();
// 如果该 channel 是 SocketChannel
if (targetChannel instanceof SocketChannel) {
// 将读到的内容写到该 channel 中
SocketChannel dest = (SocketChannel) targetChannel;
dest.write(charset.encode(content));
}
}
}
}
}
}
}
public static void main(String[] args) throws IOException {
new NServer().init();
}
}

复制代码

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Scanner;
/**
* 聊天室客户端
*/
public class NClient {
private Selector selector = null;
static final int PORT = 30000;
private Charset charset = Charset.forName("UTF-8");
private SocketChannel sc = null;
public void init() throws IOException {
selector = Selector.open();
InetSocketAddress isa = new InetSocketAddress("127.0.0.1", PORT);
// 打开套接字通道并将其连接到远程地址
sc = SocketChannel.open(isa);
// 设置为非阻塞模式
sc.configureBlocking(false);
// 注册到 selector
sc.register(selector, SelectionKey.OP_READ);
new ClientThread().start();
// 创建键盘输入流
Scanner scan = new Scanner(System.in);
while (scan.hasNextLine()) {
String line = scan.nextLine();
// 将键盘大忽如的内容输出到 SocketChannel 中
sc.write(charset.encode(line));
}
}
private class ClientThread extends Thread {
@Override
public void run() {
try {
while (selector.select() > 0) {
for (SelectionKey sk : selector.selectedKeys()) {
// 从 set集合删除正在处理的 SelectionKey
selector.selectedKeys().remove(sk);
// 如果 sk 对应的 channel 中有可读数据
if (sk.isReadable()) {
// 使用 NIO 读取 channel 中的数据
SocketChannel sc = (SocketChannel) sk.channel();
ByteBuffer buff = ByteBuffer.allocate(1024);
String content = "";
while (sc.read(buff) > 0) {
sc.read(buff);
buff.flip();
content += charset.decode(buff);
}
System.out.println("聊天信息： " + content);
// 为下一次读取做准备
sk.interestOps(SelectionKey.OP_READ);
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws IOException {
new NClient().init();
}
}

复制代码