废话不多说 直接上代码
客户端
public class ChatClient {
public static void main(String[] args) {
try {
new ChatClient().startClient(InetAddress.getLocalHost().getHostAddress());
} catch (IOException e) {
e.printStackTrace();
}
}
public void startClient(String name) throws IOException {
// 1、连接服务端
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1",8000));
// 2、接收服务器响应数据
Selector selector = Selector.open();
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
// 3、创建线程
new Thread(new ClientThread(selector)).start();
// 4、向服务器端发送消息
Scanner scanner = new Scanner(System.in);
while (scanner.hasNextLine()) {
String msg = scanner.nextLine();
if (msg.length() > 0) {
socketChannel.write(Charset.forName("UTF-8").encode(name + ": " + msg));
}
}
}
}
客户端用到的线程类
public class ClientThread implements Runnable{
private Selector selector;
public ClientThread(Selector selector) {
this.selector = selector;
}
@Override
public void run() {
try {
// while (true) {}
for(;;){
// 获取 channel 数量
int readChannels = selector.select();
if (readChannels == 0){
System.out.println("没有获取到可用连接");
continue;
}
// 获取可用 channel
Set selectionKeys = selector.selectedKeys();
// 遍历集合
Iterator iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey selectionKey = iterator.next();
// 移出 Set 集合当前 selectionKey 因为已经取出
iterator.remove();
// 6、根据就绪状态、调用对应方法实现具体业务操作
// 6.2、如果可读状态
if (selectionKey.isReadable()) {
// do something
readOperator(selector,selectionKey);
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
private void readOperator(Selector selector, SelectionKey selectionKey) throws IOException {
// 1、从 selectionKey 获取已就绪的通道
SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
// 2、创建 buffer
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
// 3、循环读取客户端消息
int readLength = socketChannel.read(byteBuffer);
String message = "";
if (readLength > 0) {
// 切换模式
byteBuffer.flip();
// 读取内容
message += Charset.forName("UTF-8").decode(byteBuffer);
}
// 4、channel 注册到 selector 上、监听可读状态
socketChannel.register(selector,selectionKey.OP_READ);
if (message.length() > 0) {
System.out.println("广播：" + message);
}
}
}
服务器
public class ChatServer {
public static void main(String[] args) {
try {
new ChatServer().startServer();
} catch (IOException e) {
e.printStackTrace();
}
}
public void startServer() throws IOException {
// 1、创建 selector 选择器
Selector selector = Selector.open();
// 2、创建 serverSocketChannel 通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 3、绑定监听端口
serverSocketChannel.bind(new InetSocketAddress(8000));
// 设置非阻塞模式
serverSocketChannel.configureBlocking(false);
// 4、把 channel 通道注册到 selector 选择器上
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("服务器启动成功~");
// 5、循环、等待新链接接入
// while (true) {}
for(;;){
// 获取 channel 数量
int readChannels = selector.select();
if (readChannels == 0){
System.out.println("没有获取到可用连接");
continue;
}
// 获取可用 channel
Set selectionKeys = selector.selectedKeys();
// 遍历集合
Iterator iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey selectionKey = iterator.next();
// 移出 Set 集合当前 selectionKey 因为已经取出
iterator.remove();
// 6、根据就绪状态、调用对应方法实现具体业务操作
// 6.1、如果 accept(接收) 状态
if (selectionKey.isAcceptable()){
// do something
acceptOperator(serverSocketChannel,selector);
}
// 6.2、如果可读状态
if (selectionKey.isReadable()) {
// do something
readOperator(selector,selectionKey);
}
}
}
}
private void readOperator(Selector selector, SelectionKey selectionKey) throws IOException {
// 1、从 selectionKey 获取已就绪的通道
SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
// 2、创建 buffer
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
// 3、循环读取客户端消息
int readLength = socketChannel.read(byteBuffer);
String message = "";
if (readLength > 0) {
// 切换模式
byteBuffer.flip();
// 读取内容
message += Charset.forName("UTF-8").decode(byteBuffer);
}
// 4、channel 注册到 selector 上、监听可读状态
socketChannel.register(selector,selectionKey.OP_READ);
// 5、把客户端发送消息、广播到其他客户端
if (message.length() > 0) {
// 广播给其他客户端
System.out.println("广播：" + message);
casrOtherClient(message,selector,socketChannel);
}
}
private void casrOtherClient(String message, Selector selector, SocketChannel socketChannel) throws IOException {
// 1、获取所有已经接入客户端
Set selectionKeySet = selector.keys();
// 2、循环向所有 channel 广播消息
for (SelectionKey selectionKey : selectionKeySet) {
// 获取每个 channel
Channel tarChannel = selectionKey.channel();
// 不需要给自己发送
if (tarChannel instanceof SocketChannel && tarChannel != socketChannel) {
((SocketChannel) tarChannel).write(Charset.forName("UTF-8").encode(message));
}
}
}
private void acceptOperator(ServerSocketChannel serverSocketChannel, Selector selector) throws IOException {
// 1、接入状态、创建 SocketChannel
SocketChannel socketChannel = serverSocketChannel.accept();
// 2、设置非阻塞模式
socketChannel.configureBlocking(false);
// 3、把 SocketChannel 注册到 selector 选择器上、监听可读状态
socketChannel.register(selector,SelectionKey.OP_READ);
// 4、客户端回复信息
socketChannel.write(Charset.forName("UTF-8").encode("已进入聊天室、请注意隐私安全！"));
}
}
