客户端:
package com.wcc.a_tcpnio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Scanner;
/**
* 符合TCP协议,非阻塞IO NIO完成对应的客户端代码
* @Author kk
* @Date 2020/3/16 15:10
*/
public class TcpNioClient {
public static void main(String[] args) throws IOException, InterruptedException {
//1.得到一个网络通道
SocketChannel socket = SocketChannel.open();
//2.设置当前IO采用的方式为非阻塞方式
socket.configureBlocking(false);
//3.确定服务器IP地址和对应程序的端口号
InetSocketAddress address = new InetSocketAddress("192.168.X.XXX",8848);
//4.连接服务器
if(!socket.connect(address)){
//如果是false,表示连接失败,保持申请连接的状态
while(!socket.finishConnect()){
Thread.sleep(2000);
//因为采用NIO非阻塞方式,再获取等待连接的状态下,可以去做当前程序的其他操作
System.out.println("保持呼叫服务器状态,但是我还能做点别的事情,等待2秒继续申请链接");
}
}
//5.准备一个数据存入到缓冲区
ByteBuffer buffer = ByteBuffer.wrap("你好服务器,我在等你".getBytes());
//6.通过SocketChannel符合TCP协议Socket要求Channel对象发送
socket.write(buffer);
new Scanner(System.in).nextLine();
}
}
服务端:
1.开启服务
ServerScoketChannel
2.开启Selector大哥
Selector对象
3.服务器ServerSocketChannel bind监听端口号
8848端口
4.设置非阻塞状态
configureBlocking(false);
5.ServerSocketChannel 注册到 Selector 返回值是一个SelectionKey
register(selector, OP_ACCEPT)
6.Selector大哥开始忙活
6.1 获取连接,注册对应的Socket
6.2 监听读写事件
6.2.1 读数据,客户端发送数据到服务器
6.2.2 写数据,发送数据给客户端
package com.wcc.a_tcpnio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
/**
* 使用ServerSocketChannel NIO非阻塞方式,完成服务端代码
* @Author kk
* @Date 2020/3/16 15:44
*/
public class TcpNioServer {
public static void main(String[] args) throws IOException {
//1.开启服务器
ServerSocketChannel serverSocket = ServerSocketChannel.open();
//2.开启Selector
Selector selector = Selector.open();
//3.服务端代码绑定端口号
serverSocket.bind(new InetSocketAddress(8848));
//4.设定非阻塞状态
serverSocket.configureBlocking(false);
//5.ServerSocketChannel 注册到 Selector 返回值是一个SelectionKey
//并且明确当前Selector监听SelectionKey.OP_ACCEPT,监听服务器
serverSocket.register(selector, SelectionKey.OP_ACCEPT);
//6.大哥干活
while(true){
//6.1获取连接,注册对应的Socket
if(0 == selector.select(1000)){
//表示没有连接到客户端
System.out.println("ServerSocket提示:当前没有客户端搭理我");
continue;
}
//6.2监听读写事件
//得到当前Selector中所有的SelectionKey
Iterator selectionKeys = selector.selectedKeys().iterator();
while (selectionKeys.hasNext()){
SelectionKey selectionKey = selectionKeys.next();
// 6.2.1 判断客户端是一个连接请求 OP_ACCEPT
if(selectionKey.isAcceptable()){
System.out.println("客户端请求连接");
//获取对应的Socket,只不过这里是获取对应的SocketChannel
SocketChannel socket = serverSocket.accept();
//设置当前对应客户端的SocketChannel对象是一个非阻塞状态
socket.configureBlocking(false);
/*
注册当前Socket对象
selector注册到当前的Selector【核心】
SelectionKey.OP_READ 选择当前Scoket监听的操作内容是OP_READ 从当前Socket中读取数据
ByteBuffer.allocate(1024*4) attachment补充参数,这里是给予当前Socket对象一个4kb字节缓冲区对象
*/
socket.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024*4));
}
//6.2.2 判断客户端目前是可读状态,获取客户端发送给服务器的数据
if(selectionKey.isReadable()){
//从SelectionKey中获取对应的SocketChannel对象
SocketChannel socket = (SocketChannel)selectionKey.channel();
//因为使用的是NIO,涉及到Channel和ByteBuffer,数据在缓冲区中
ByteBuffer buffer = (ByteBuffer)selectionKey.attachment();
//利用SocketChannel从ByteBuffer缓冲中读取数据
int read = socket.read(buffer);
System.out.println("客户端发送数据:" + new String(buffer.array()));
}
//处理完进行一个一处当前SelcetionKey操作
selectionKeys.remove();
}
}
}
}
NIO TCP 聊天室客户端完成
package com.wcc.b_niochat;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
/**
* NIO非阻塞状态的TCP聊天室客户端核心代码
* @Author kk
* @Date 2020/3/16 16:21
*/
public class ChatClient {
/**
* 服务器IP地址
*/
private static final String HOST = "192.168.2.198";
/**
* 服务器连接对应的端口号
*/
private static final int PORT = 8848;
/**
* 符合NIO要求的SocketChannel对象
*/
private SocketChannel socket;
/**
* 用户名
*/
private String userName;
/**
* 客户端构造方法,创建客户端对象
* @param userName 准备的用户名
* @throws IOException
* @throws InterruptedException
*/
public ChatClient(String userName) throws IOException, InterruptedException {
//1.打开SocketChannel
socket = SocketChannel.open();
//2.设置非阻塞状态
socket.configureBlocking(false);
//3.根据指定的HOST IP地址和对应的PORT端口号创建对应的InetSocketAddress
InetSocketAddress address = new InetSocketAddress(HOST, PORT);
//4.连接服务器
if(!socket.connect(address)){
//如果没有连接到服务器,保持请求连接的状态
while (!socket.finishConnect()){
System.out.println("服务器请求连接失败,等待2秒继续请求连接");
Thread.sleep(2000);
}
}
this.userName = userName;
System.out.println("客户端 " + userName + "准备就绪");
}
/*
这里需要完成两个方法
一个是发送数据到服务器
一个是接收服务器发送的数据
*/
/**
* 发送数据到服务器,用于广播消息,群聊
* @param message 指定的消息
*/
public void sendMsg(String message) throws IOException {
//断开服务器连接 close
if("close".equals(message)){
socket.close();
return;
}
/*
StringBuffer 线程安全,效率低
StringBuilder 线程不安全,效率高
*/
message = userName + ":" +message;
ByteBuffer buffer = ByteBuffer.wrap(message.getBytes());
socket.write(buffer);
}
public void receiveMsg() throws IOException {
//准备ByteBuffer
ByteBuffer buffer = ByteBuffer.allocate(1024 * 4);
int length = socket.read(buffer);
if(length > 0){
System.out.println(new String(buffer.array(), 0 , length));
}
}
}
NIO TCP聊天室服务端完成
package com.wcc.b_niochat;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
/**
* NIO非阻塞状态的TCP聊天室服务端核心代码
*
* @Author kk
* @Date 2020/3/16 17:00
*/
public class ChatServer {
/**
* 服务器核心模块,ServerSocketChannel
*/
private ServerSocketChannel serverSocket;
/**
* 服务端NIO Selector选择器
*/
private Selector selector;
/**
* 服务端监听服务指定端口号
*/
private static final int PORT = 8848;
/*
构造方法
接受方法
发送发放(广播)
同时启动接收和发送功能 start
*/
/**
* 服务器启动构造方法,开启ServerSocket,同时开启Selector注册操作
* @throws IOException 异常
*/
public ChatServer() throws IOException {
//1.启动服务器ServerSocket NIO服务器
serverSocket = ServerSocketChannel.open();
//2.启动选择器
selector = Selector.open();
//3.端口绑定
serverSocket.bind(new InetSocketAddress(PORT));
//4.选择NIO方式为非阻塞状态
serverSocket.configureBlocking(false);
//5.注册ServerSocket,确定当前监听的状态是OP_ACCEPT
serverSocket.register(selector, SelectionKey.OP_ACCEPT);
}
/**
* 服务器干活的方法,执行:
* 指定客户端绑定
* 数据接收和转发
*
*/
public void start() {
try {
while (true) {
if (0 == selector.select(2000)) {
System.out.println("服务器默默的等待连接...");
continue;
}
/*
selectedKeys:
获取当前所有发生事件操作的对应SelectionKey Set集合
*/
Iterator iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
//1.连接
if (key.isAcceptable()) {
//连接客户端,获取对应的SocketChannel对象
SocketChannel socket = serverSocket.accept();
socket.configureBlocking(false);
socket.register(selector, SelectionKey.OP_READ);
//广播用户上线
broadcast(socket, socket.getRemoteAddress().toString() + "上线了");
}
//2.接收数据转发
if (key.isReadable()) {
readMessage(key);
}
iterator.remove();
}
}
} catch (IOException e){
e.printStackTrace();
}
}
/**
* 从指定的SelectionKey中读取数据
* @param key
*/
public void readMessage(SelectionKey key) throws IOException {
//1.根据指定的SelectionKey获取对应的SelectableChannel对象
SocketChannel socket = (SocketChannel)key.channel();
//2.创建缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
int length = socket.read(buffer);
if(length > 0){
String message = new String(buffer.array());
//广播数据
}
}
/**
* 广播方法,该方法是群发消息,但是不发给自己
* @param self 当前发送数据的客户端
* @param message 要发送的消息
*/
public void broadcast(SocketChannel self, String message) throws IOException {
//获取当前Selector选择器中所有的SelectionKey
Set keys = selector.keys();
//遍历整个SelectionKey Set集合
for (SelectionKey key : keys) {
//获取对应SelectionKey的channel对象
SelectableChannel channel = key.channel();
//channel对应的是一个SocketChannel对象,并且不是当前发送消息的SocketChannel对象
if(channel instanceof SocketChannel && !channel.equals(self)){
SocketChannel socketChannel = (SocketChannel) channel;
//根据指定的Byte类型数组,创建对应的ByteBuffer缓冲区
ByteBuffer buffer = ByteBuffer.wrap(message.getBytes());
socketChannel.write(buffer);
}
}
}
public static void main(String[] args) throws IOException {
new ChatServer().start();
}
}
NIO TCP聊天室客户端线程代码实现
package com.wcc.b_niochat;
import java.io.IOException;
import java.util.Scanner;
/**
* 客户端线程
* @Author kk
* @Date 2020/3/16 17:37
*/
public class ChatClientThread {
public static void main(String[] args) throws IOException, InterruptedException {
Scanner scanner = new Scanner(System.in);
System.out.println("请输入用户名:");
String userName = scanner.nextLine();
scanner.nextLine();
if(0 == userName.length()){
return;
}
ChatClient chatClient = new ChatClient(userName);
//接收消息
new Thread(() ->{
while(true){
try {
chatClient.receiveMsg();
} catch (IOException e) {
e.printStackTrace();
}
}
}).start();
//发送消息
while(scanner.hasNextLine()){
String msg = scanner.nextLine();
chatClient.sendMsg(msg);
}
}
}
作者:申士_