码云地址
1.服务器代码
public class NIOServer {
/**
* 用户列表
* key: 用户名(唯一)
* value: 此用户与服务器建立的连接Channel
*/
Map<String, SocketChannel> users;
/**
* 选择器
*/
private Selector selector;
/**
* 服务器端口
*/
private static final int PORT = 8866;
/**
* 服务器
*/
private ServerSocketChannel server;
/**
* 服务器缓冲区
*/
private ByteBuffer serverBuffer;
/**
* 解码器
*/
private CharsetDecoder decoder;
/**
* 编码器
*/
private CharsetEncoder encoder;
public void setServer() throws IOException {
server = ServerSocketChannel.open();
//设置server为非阻塞模式
serverBuffer = ByteBuffer.allocate(1024);
server.socket().bind(new InetSocketAddress(PORT));
server.configureBlocking(false);
selector = Selector.open();
//服务器监听accept()事件
server.register(selector, SelectionKey.OP_ACCEPT);
}
public NIOServer() throws IOException {
setServer();
//初始化用户列表
users = new ConcurrentHashMap<>();
decoder = Charset.forName("UTF-8").newDecoder();
encoder = Charset.forName("UTF-8").newEncoder();
}
public void startServer() throws IOException {
System.out.println("***** 服务器开始工作 *****");
while (true) {
selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
//获取客户端连接
SocketChannel clientChannel = server.accept();
clientChannel.configureBlocking(false);
//分配一个buffer
ByteBuffer clientBuffer = ByteBuffer.allocate(1024);
Information channelInformation = new Information(null, clientBuffer);
/*
* @args1 注册到selector上
* @args2 监听事件
* @args3 为每一个客户端连接Channel绑定一个Information
* */
clientChannel.register(selector, SelectionKey.OP_READ, channelInformation);
System.out.println("---------------------------------");
String s = clientChannel.socket().getInetAddress().toString();
System.out.print("获取连接IP： " + s.substring(1, s.length()) + " 端口：");
System.out.println(clientChannel.socket().getPort());
} else if (key.isReadable()) {
SocketChannel clientChannel = (SocketChannel) key.channel();
try {
Information information = (Information) key.attachment();
ByteBuffer clientBuffer = information.getBuffer();
clientBuffer.clear();
int n = clientChannel.read(clientBuffer);
//返回-1，表示客户端正常断开连接
if (n == -1) {
String username = information.getUsername();
//释放buffer
information.release();
key.cancel();
clientChannel.close();
System.out.println("---------------------------------");
System.out.println("用户: " + username + " 已下线");
System.out.println("---------------------------------");
} else {
clientBuffer.flip();
String message = decoder.decode(clientBuffer).toString();
//规定 给服务器的消息前缀为 To Sever:username
//给其他客户端的消息前缀为 To {username}: xxxx
if (message.startsWith("To Server:")) {
String username = message.substring(10, message.length());
information.setUsername(username);
users.put(username, clientChannel);
System.out.println("用户名： " + username);
System.out.println("---------------------------------");
} else {
//转发给其他客户端的消息
String[] data = message.split("\\:");
String fromUsername = information.getUsername();
String toUsername = data[0].substring(3, data[0].length());
String content = data[1];
System.out.println(fromUsername + "——>" + toUsername + ": " + content);
if (users.containsKey(toUsername)) {
SocketChannel toChannel = users.get(toUsername);
toChannel.write(encoder.encode(CharBuffer.wrap(("收到来自用户" + fromUsername + "的消息" + ": " + content))));
}
}
}
} catch (Exception e) {
key.cancel();
clientChannel.close();
System.out.println("客户端异常断开。。。 服务器关闭连接，客户端IP地址为： " + clientChannel.socket().getInetAddress() + " 的连接");
}
}
}
}
}
public static void main(String[] args) throws IOException {
new NIOServer().startServer();
}
}
2.附件Information类
import java.nio.ByteBuffer;
@Data
public class Information {
private String username;
private ByteBuffer buffer;
public Information(String username, ByteBuffer buffer) {
this.username = username;
this.buffer = buffer;
}
public void release() {
this.buffer = null;
}
}
3.客户端
//继承Thread，方便测试时不断输入数据
public class NIOClient extends Thread {
/**
* 解码器
*/
private CharsetDecoder decoder;
/**
* 编码器
*/
private CharsetEncoder encoder;
/**
* 用户名
*/
private String username;
/**
* 选择器
*/
private Selector selector;
/**
* 客户端
*/
private SocketChannel clientChannel;
/**
* 缓冲区
*/
private ByteBuffer buffer;
public NIOClient(String username) throws IOException {
decoder = Charset.forName("UTF-8").newDecoder();
encoder = Charset.forName("UTF-8").newEncoder();
buffer = ByteBuffer.allocate(1024);
this.username = username;
clientChannel = SocketChannel.open();
clientChannel.configureBlocking(false);
selector = Selector.open();
clientChannel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
clientChannel.connect(new InetSocketAddress("localhost", 8866));
}
@Override
public void run() {
while (true) {
try {
selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (key.isConnectable()) {
SocketChannel channel = (SocketChannel) key.channel();
//这里一定要加上，保证完全客户端与服务器完全建立连接之后才发送数据
if (channel.isConnectionPending()) {
channel.finishConnect();
}
send("To Server:" + username);
} else {
SocketChannel channel = (SocketChannel) key.channel();
channel.read(buffer);
buffer.flip();
String msg = decoder.decode(buffer).toString();
buffer.clear();
System.out.println(msg);
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 向服务器发送消息
*/
public void send(String msg) throws IOException {
this.clientChannel.write(encoder.encode(CharBuffer.wrap(msg)));
}
/**
* 关闭连接
*/
public void close() throws IOException {
selector.close();
clientChannel.socket().close();
}
}
4.测试类
//测试时，复制此代码，开启多个客户端进行测试。
public class Client {
public static void main(String[] args) throws IOException {
String username = "XXX";
NIOClient client = new NIOClient(username);
client.start();
// 输入输出流
BufferedReader sin = new BufferedReader(new InputStreamReader(System.in));
try {
// 循环读取键盘输入
String readline;
while ((readline = sin.readLine()) != null) {
if (readline.equals("bye")) {
client.close();
System.exit(0);
}
// 发送消息
client.send(readline);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
5.测试结果
先启动服务器，然后启动两个客户端，一个用户名为kd7，另一个为shstart7
服务器控制台
客户端shstart7控制台
客户端kd7控制台
6.运行流程
- 1.初始化服务器，包括绑定注册Selector，监听accept()事件等，然后开启服务器。
- 2.客户端
(绑定了客户端自己的Slector，监听Connect和Read事件)连接服务器，服务器监听到连接，获取SocketChannel，将其注册到Selector上，监听read()事件，并打印客户端IP地址与端口号。 - 3.客户端监听到Connect事件，向服务器发送自己的用户名
(固定格式 To Server:username)。 - 4.服务器监听到read()事件，根据前缀To Server判断是用户发来的用户名，然后获取用户名，将
<username：SocketChannel>存到Map中，然后申请一个ByteBuffer，将获取到的username和ByteBuffer封装成一个Information，以附件的形式添加到此连接SocketChannel中。 - 5.客户端A向客户端B发送消息。
- 固定格式：To usernameB: msg，服务器监听到read()事件，将消息读取到自己绑定的buffer中，解析出消息然后根据前缀判断此消息是发给其他客户端的，获取消息中B的username，查询Map获取B的SocketChannel，然后将消息发送给B。服务器打印一下A——>B : msg
