NIO 优势在于使用了Selector/Channel，Selector可以根据注册到其中的channel，判断key，来执行不同的事件。
服务端：
- 创建ServerSocketChannel, 并设置为非阻塞模式
- 绑定端口
- 获取Selector，并将ServerSocketChannel注册到Selector中
- 获取客户端读取事件，将读取的内容分发到其他客户端
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
/**
* 群聊转发
*/
public class Server {
private Selector selector;
private ServerSocketChannel ssChannel;
private static final int PORT = 8888;
public Server() {
try {
ssChannel = ServerSocketChannel.open();
ssChannel.configureBlocking(false);
ssChannel.bind(new InetSocketAddress(PORT));
selector = Selector.open();
ssChannel.register(selector, SelectionKey.OP_ACCEPT);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
public void listen() {
try {
while (selector.select() > 0) {
System.out.println("selector准备就绪");
// 获取当前选择器种所有注册的“事件key”
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey sk = iterator.next();
if (sk.isAcceptable()) {
System.out.println("selector有数据 isAcceptable:" + sk.isAcceptable());
// 准备就绪，则获取客户端连接
SocketChannel channel = ssChannel.accept();
// 切换为非阻塞
channel.configureBlocking(false);
// 将该通道注册到选择器上
channel.register(selector, SelectionKey.OP_READ);
} else if (sk.isReadable()) {
System.out.println("selector有数据 isReadable:" + sk.isReadable());
// 转发
readClientData(sk);
}
// 移除该事件
iterator.remove();
}
}
} catch (Exception e) {
System.out.println("socket处理异常");
e.printStackTrace();
}
}
private void readClientData(SelectionKey sk) throws IOException {
SocketChannel sChannel = null;
try {
sChannel = (SocketChannel) sk.channel();
// 读取
ByteBuffer buffer = ByteBuffer.allocate(1024);
int len = sChannel.read(buffer);
if (len > 0) {
buffer.flip();
// 转发到其他客户端
String msg = new String(buffer.array(), 0, len);
System.out.println("接收到客户端消息"+ msg);
sendMsgToAll(msg, sChannel);
}
} catch (Exception e) {
try {
System.out.println("用户下线" + sChannel.getRemoteAddress());
sk.cancel();
sChannel.close();
}catch (Exception ex) {
System.out.println("关闭客户端");
}
e.printStackTrace();
}
}
private void sendMsgToAll(String buffer, SocketChannel sChannel) throws IOException {
// 获取全部的在线channel
Set<SelectionKey> keys = selector.keys();
for (SelectionKey selectionKey : keys) {
// 获取channel
Channel channel = selectionKey.channel();
if (channel instanceof SocketChannel && channel != sChannel) {
// 缓冲区
ByteBuffer msg = ByteBuffer.wrap(buffer.getBytes());
((SocketChannel)channel).write(msg);
}
}
}
public static void main(String[] args) throws IOException {
Server server = new Server();
server.listen();
}
}
客户端：
- 获取Selector，
- 获取SocketChannel
- 将SocketChannel注册到Selector
- selector监听OP_READ事件
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.util.Iterator;
import java.util.Scanner;
public class Client {
private Selector selector;
private SocketChannel sChannel;
private static final int PORT = 8888;
private String clientName = "client_1";
public Client() {
try {
selector = Selector.open();
sChannel = SocketChannel.open(new InetSocketAddress(PORT));
sChannel.configureBlocking(false);
sChannel.register(selector, SelectionKey.OP_READ);
System.out.println(Thread.currentThread().getName() + "启动成功");
} catch (IOException e) {
e.printStackTrace();
}
}
public Client(String name) {
try {
clientName = name;
selector = Selector.open();
sChannel = SocketChannel.open(new InetSocketAddress(PORT));
sChannel.configureBlocking(false);
sChannel.register(selector, SelectionKey.OP_READ);
System.out.println(Thread.currentThread().getName() + "启动成功");
} catch (IOException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws IOException {
Client client = new Client("client_2");
new Thread(new Runnable() {
@Override
public void run() {
try {
client.readInfo();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}).start();
Scanner scanner = new Scanner(System.in);
while (scanner.hasNextLine()) {
String str = scanner.nextLine();
client.sendMessage2Server(client.clientName, str);
}
}
private void sendMessage2Server(String clientName, String msg) throws IOException {
sChannel.write(ByteBuffer.wrap((clientName + ": " + msg).getBytes()));
}
private void readInfo() throws IOException {
while (selector.select() > 0) {
// 获取当前选择器种所有注册的“事件key”
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey sk = iterator.next();
if (sk.isReadable()) {
// 获取选择器上 读就绪 的通道
SocketChannel sChannel = (SocketChannel) sk.channel();
// 读取
ByteBuffer buffer = ByteBuffer.allocate(1024);
sChannel.read(buffer);
System.out.println(new String(buffer.array()).trim());
}
// 移除该事件
iterator.remove();
}
}
}
}
