mirror of
https://gitee.com/chinabugotech/hutool.git
synced 2025-04-19 03:01:48 +08:00
nio
This commit is contained in:
hyl
parent
baa8ddd9ed
commit
28c9801566
@ -2,12 +2,20 @@ package cn.hutool.socket.nio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.core.thread.ThreadFactoryBuilder;
|
||||
|
||||
import java.io.Closeable;
|
||||
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.ServerSocketChannel;
|
||||
import java.nio.channels.SocketChannel;
|
||||
import java.util.Iterator;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
|
||||
/**
|
||||
* NIO客户端
|
||||
@ -15,9 +23,11 @@ import java.nio.channels.SocketChannel;
|
||||
* @author looly
|
||||
* @since 4.4.5
|
||||
*/
|
||||
public class NioClient implements Closeable {
|
||||
public abstract class NioClient implements Closeable {
|
||||
|
||||
private Selector selector;
|
||||
private SocketChannel channel;
|
||||
private ExecutorService executorService;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
@ -46,13 +56,111 @@ public class NioClient implements Closeable {
|
||||
*/
|
||||
public NioClient init(InetSocketAddress address) {
|
||||
try {
|
||||
this.channel = SocketChannel.open(address);
|
||||
//创建一个SocketChannel对象，配置成非阻塞模式
|
||||
this.channel = SocketChannel.open();
|
||||
channel.configureBlocking(false);
|
||||
|
||||
//创建一个选择器，并把SocketChannel交给selector对象
|
||||
this.selector = Selector.open();
|
||||
channel.register(selector, SelectionKey.OP_CONNECT);
|
||||
|
||||
//发起建立连接的请求，这里会立即返回，当连接建立完成后，SocketChannel就会被选取出来
|
||||
channel.connect(address);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查连接是否建立完成
|
||||
*/
|
||||
public boolean waitConnect() throws IOException {
|
||||
boolean isConnect = false;
|
||||
while (0 != this.selector.select()) {
|
||||
final Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
|
||||
while (keyIter.hasNext()) {
|
||||
//连接建立完成
|
||||
SelectionKey key = keyIter.next();
|
||||
if (key.isConnectable()) {
|
||||
if (this.channel.finishConnect()) {
|
||||
this.channel.register(selector, SelectionKey.OP_READ);
|
||||
isConnect = true;
|
||||
}
|
||||
}
|
||||
keyIter.remove();
|
||||
break;
|
||||
}
|
||||
if (isConnect) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return isConnect;
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*/
|
||||
public void listen() {
|
||||
this.executorService = Executors.newSingleThreadExecutor(r -> {
|
||||
final Thread thread = Executors.defaultThreadFactory().newThread(r);
|
||||
thread.setName("nio-client-listen");
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
});
|
||||
this.executorService.execute(() -> {
|
||||
try {
|
||||
doListen();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
private void doListen() throws IOException {
|
||||
while (0 != this.selector.select()) {
|
||||
// 返回已选择键的集合
|
||||
final Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
|
||||
while (keyIter.hasNext()) {
|
||||
handle(keyIter.next());
|
||||
keyIter.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理SelectionKey
|
||||
*
|
||||
* @param key SelectionKey
|
||||
*/
|
||||
private void handle(SelectionKey key) throws IOException {
|
||||
//连接建立完成
|
||||
// if (key.isConnectable()) {
|
||||
// if (this.channel.finishConnect()) {
|
||||
// this.channel.register(selector, SelectionKey.OP_READ);
|
||||
// }
|
||||
// }
|
||||
|
||||
// 读事件就绪
|
||||
if (key.isReadable()) {
|
||||
final SocketChannel socketChannel = (SocketChannel) key.channel();
|
||||
read(socketChannel);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理读事件<br>
|
||||
* 当收到读取准备就绪的信号后，回调此方法，用户可读取从客户端传出来的消息
|
||||
*
|
||||
* @param socketChannel SocketChannel
|
||||
*/
|
||||
protected abstract void read(SocketChannel socketChannel);
|
||||
|
||||
/**
|
||||
* 处理读事件<br>
|
||||
* 当收到读取准备就绪的信号后，回调此方法，用户可读取从客户端传世来的消息
|
||||
@ -85,18 +193,14 @@ public class NioClient implements Closeable {
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取SocketChannel
|
||||
*
|
||||
* @return SocketChannel
|
||||
* @since 5.3.10
|
||||
*/
|
||||
public SocketChannel getChannel() {
|
||||
return this.channel;
|
||||
public void closeListen() {
|
||||
this.executorService.shutdown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
IoUtil.close(this.selector);
|
||||
IoUtil.close(this.channel);
|
||||
closeListen();
|
||||
}
|
||||
}
|
||||
|
||||
@ -137,7 +137,7 @@ public abstract class NioServer implements Closeable {
|
||||
|
||||
/**
|
||||
* 处理读事件<br>
|
||||
* 当收到读取准备就绪的信号后，回调此方法，用户可读取从客户端传世来的消息
|
||||
* 当收到读取准备就绪的信号后，回调此方法，用户可读取从客户端传出来的消息
|
||||
*
|
||||
* @param socketChannel SocketChannel
|
||||
*/
|
||||
|
||||
@ -0,0 +1,95 @@
|
||||
package cn.hutool.socket;
|
||||
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import cn.hutool.socket.nio.NioClient;
|
||||
import lombok.SneakyThrows;
|
||||
|
||||
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.Set;
|
||||
|
||||
public class NioClientTest {
|
||||
|
||||
@SneakyThrows
|
||||
public static void main(String[] args) {
|
||||
NioClient client = new NioClient("127.0.0.1", 8080) {
|
||||
@SneakyThrows
|
||||
@Override
|
||||
protected void read(SocketChannel sc) {
|
||||
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
|
||||
//从channel读数据到缓冲区
|
||||
int readBytes = sc.read(readBuffer);
|
||||
if (readBytes > 0){
|
||||
//Flips this buffer. The limit is set to the current position and then
|
||||
// the position is set to zero，就是表示要从起始位置开始读取数据
|
||||
readBuffer.flip();
|
||||
//eturns the number of elements between the current position and the limit.
|
||||
// 要读取的字节长度
|
||||
byte[] bytes = new byte[readBuffer.remaining()];
|
||||
//将缓冲区的数据读到bytes数组
|
||||
readBuffer.get(bytes);
|
||||
String body = new String(bytes, "UTF-8");
|
||||
System.out.println("the read client receive message: " + body);
|
||||
doWrite(sc, body);
|
||||
}else if(readBytes < 0){
|
||||
sc.close();
|
||||
}
|
||||
}
|
||||
};
|
||||
if (client.waitConnect()) {
|
||||
client.listen();
|
||||
}
|
||||
ByteBuffer buffer = ByteBuffer.wrap("client 发生到 server".getBytes());
|
||||
client.write(buffer);
|
||||
if(!buffer.hasRemaining()) {
|
||||
System.err.println("第一次发送成功...");
|
||||
}
|
||||
buffer = ByteBuffer.wrap("client 再次发生到 server".getBytes());
|
||||
client.write(buffer);
|
||||
if(!buffer.hasRemaining()) {
|
||||
System.err.println("第二次发送成功...");
|
||||
}
|
||||
}
|
||||
|
||||
//发送请求
|
||||
private static void doWriteRequest(SocketChannel socketChannel) throws Exception{
|
||||
System.err.println("start connect...");
|
||||
|
||||
//创建ByteBuffer对象，会放入数据
|
||||
ByteBuffer byteBuffer = ByteBuffer.allocate("Hello nio.example.Server!".getBytes().length);
|
||||
byteBuffer.put("Hello nio.example.Server!".getBytes());
|
||||
byteBuffer.flip();
|
||||
//写数据
|
||||
socketChannel.write(byteBuffer);
|
||||
if(!byteBuffer.hasRemaining()) {
|
||||
System.err.println("Send request success...");
|
||||
}
|
||||
}
|
||||
|
||||
//读取服务端的响应
|
||||
private static void doRead(SelectionKey selectionKey) throws Exception{
|
||||
SocketChannel socketChannel = ((SocketChannel) selectionKey.channel());
|
||||
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
|
||||
int len = socketChannel.read(byteBuffer);
|
||||
System.out.println("Recv:" + new String(byteBuffer.array(), 0 ,len));
|
||||
}
|
||||
|
||||
public static void doWrite(SocketChannel channel, String response) throws IOException {
|
||||
response = "我们已收到消息："+response;
|
||||
if(!StrUtil.isBlank(response)){
|
||||
byte [] bytes = response.getBytes();
|
||||
//分配一个bytes的length长度的ByteBuffer
|
||||
ByteBuffer write = ByteBuffer.allocate(bytes.length);
|
||||
//将返回数据写入缓冲区
|
||||
write.put(bytes);
|
||||
write.flip();
|
||||
//将缓冲数据写入渠道，返回给客户端
|
||||
channel.write(write);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,82 @@
|
||||
package cn.hutool.socket;
|
||||
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import cn.hutool.socket.nio.NioServer;
|
||||
import lombok.SneakyThrows;
|
||||
|
||||
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.ServerSocketChannel;
|
||||
import java.nio.channels.SocketChannel;
|
||||
import java.util.Set;
|
||||
|
||||
public class NioServerTest {
|
||||
|
||||
public static void main(String[] args) {
|
||||
NioServer server = new NioServer(8080) {
|
||||
@SneakyThrows
|
||||
@Override
|
||||
protected void read(SocketChannel sc) {
|
||||
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
|
||||
//从channel读数据到缓冲区
|
||||
int readBytes = sc.read(readBuffer);
|
||||
if (readBytes > 0){
|
||||
//Flips this buffer. The limit is set to the current position and then
|
||||
// the position is set to zero，就是表示要从起始位置开始读取数据
|
||||
readBuffer.flip();
|
||||
//eturns the number of elements between the current position and the limit.
|
||||
// 要读取的字节长度
|
||||
byte[] bytes = new byte[readBuffer.remaining()];
|
||||
//将缓冲区的数据读到bytes数组
|
||||
readBuffer.get(bytes);
|
||||
String body = new String(bytes, "UTF-8");
|
||||
System.out.println("the read server receive message: " + body);
|
||||
doWrite(sc, body);
|
||||
}else if(readBytes < 0){
|
||||
sc.close();
|
||||
}
|
||||
}
|
||||
|
||||
@SneakyThrows
|
||||
@Override
|
||||
protected void write(SocketChannel sc) {
|
||||
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
|
||||
//从channel读数据到缓冲区
|
||||
int readBytes = sc.read(readBuffer);
|
||||
if (readBytes > 0){
|
||||
//Flips this buffer. The limit is set to the current position and then
|
||||
// the position is set to zero，就是表示要从起始位置开始读取数据
|
||||
readBuffer.flip();
|
||||
//eturns the number of elements between the current position and the limit.
|
||||
// 要读取的字节长度
|
||||
byte[] bytes = new byte[readBuffer.remaining()];
|
||||
//将缓冲区的数据读到bytes数组
|
||||
readBuffer.get(bytes);
|
||||
String body = new String(bytes, "UTF-8");
|
||||
System.out.println("the write server receive message: " + body);
|
||||
doWrite(sc, body);
|
||||
}else if(readBytes < 0){
|
||||
sc.close();
|
||||
}
|
||||
}
|
||||
};
|
||||
server.listen();
|
||||
}
|
||||
|
||||
public static void doWrite(SocketChannel channel, String response) throws IOException {
|
||||
response = "我们已收到消息："+response;
|
||||
if(!StrUtil.isBlank(response)){
|
||||
byte [] bytes = response.getBytes();
|
||||
//分配一个bytes的length长度的ByteBuffer
|
||||
ByteBuffer write = ByteBuffer.allocate(bytes.length);
|
||||
//将返回数据写入缓冲区
|
||||
write.put(bytes);
|
||||
write.flip();
|
||||
//将缓冲数据写入渠道，返回给客户端
|
||||
channel.write(write);
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user