This commit is contained in:
Looly 2020-08-01 19:23:55 +08:00
parent 04917da6e2
commit 02b0f6194f
5 changed files with 60 additions and 57 deletions

View File

@ -6,12 +6,14 @@ import java.nio.channels.SocketChannel;
* NIO数据处理接口通过实现此接口可以从{@link SocketChannel}中读写数据 * NIO数据处理接口通过实现此接口可以从{@link SocketChannel}中读写数据
* *
*/ */
@FunctionalInterface
public interface ChannelHandler { public interface ChannelHandler {
/** /**
* 处理NIO数据 * 处理NIO数据
* *
* @param socketChannel {@link SocketChannel} * @param socketChannel {@link SocketChannel}
* @throws Exception 可能的处理异常
*/ */
void handle(SocketChannel socketChannel); void handle(SocketChannel socketChannel) throws Exception;
} }

View File

@ -3,6 +3,7 @@ package cn.hutool.socket.nio;
import cn.hutool.core.io.IORuntimeException; import cn.hutool.core.io.IORuntimeException;
import cn.hutool.core.io.IoUtil; import cn.hutool.core.io.IoUtil;
import cn.hutool.core.thread.ThreadUtil; import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.socket.SocketRuntimeException;
import java.io.Closeable; import java.io.Closeable;
import java.io.IOException; import java.io.IOException;
@ -19,9 +20,11 @@ import java.util.Iterator;
* @author looly * @author looly
* @since 4.4.5 * @since 4.4.5
*/ */
public abstract class NioClient implements Closeable { public class NioClient implements Closeable {
private Selector selector; private Selector selector;
private SocketChannel channel; private SocketChannel channel;
private ChannelHandler handler;
/** /**
* 构造 * 构造
@ -68,6 +71,17 @@ public abstract class NioClient implements Closeable {
return this; return this;
} }
/**
* 设置NIO数据处理器
*
* @param handler {@link ChannelHandler}
* @return this
*/
public NioClient setChannelHandler(ChannelHandler handler){
this.handler = handler;
return this;
}
/** /**
* 开始监听 * 开始监听
*/ */
@ -106,18 +120,14 @@ public abstract class NioClient implements Closeable {
// 读事件就绪 // 读事件就绪
if (key.isReadable()) { if (key.isReadable()) {
final SocketChannel socketChannel = (SocketChannel) key.channel(); final SocketChannel socketChannel = (SocketChannel) key.channel();
read(socketChannel); try{
handler.handle(socketChannel);
} catch (Exception e){
throw new SocketRuntimeException(e);
}
} }
} }
/**
* 处理读事件<br>
* 当收到读取准备就绪的信号后回调此方法用户可读取从客户端传出来的消息
*
* @param socketChannel SocketChannel
*/
protected abstract void read(SocketChannel socketChannel);
/** /**
* 实现写逻辑<br> * 实现写逻辑<br>
* 当收到写出准备就绪的信号后回调此方法用户可向客户端发送消息 * 当收到写出准备就绪的信号后回调此方法用户可向客户端发送消息

View File

@ -3,6 +3,7 @@ package cn.hutool.socket.nio;
import cn.hutool.core.io.IORuntimeException; import cn.hutool.core.io.IORuntimeException;
import cn.hutool.core.io.IoUtil; import cn.hutool.core.io.IoUtil;
import cn.hutool.log.Log; import cn.hutool.log.Log;
import cn.hutool.log.StaticLog;
import java.io.Closeable; import java.io.Closeable;
import java.io.IOException; import java.io.IOException;
@ -135,7 +136,12 @@ public class NioServer implements Closeable {
// 读事件就绪 // 读事件就绪
if (key.isReadable()) { if (key.isReadable()) {
final SocketChannel socketChannel = (SocketChannel) key.channel(); final SocketChannel socketChannel = (SocketChannel) key.channel();
handler.handle(socketChannel); try{
handler.handle(socketChannel);
} catch (Exception e){
IoUtil.close(socketChannel);
StaticLog.error(e);
}
} }
} }

View File

@ -1,58 +1,49 @@
package cn.hutool.socket.nio; package cn.hutool.socket.nio;
import cn.hutool.core.io.BufferUtil;
import cn.hutool.core.lang.Console; import cn.hutool.core.lang.Console;
import cn.hutool.core.util.CharsetUtil;
import cn.hutool.core.util.StrUtil; import cn.hutool.core.util.StrUtil;
import lombok.SneakyThrows; import lombok.SneakyThrows;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Scanner; import java.util.Scanner;
public class NioClientTest { public class NioClientTest {
@SneakyThrows @SneakyThrows
public static void main(String[] args) { public static void main(String[] args) {
NioClient client = new NioClient("127.0.0.1", 8080) { NioClient client = new NioClient("127.0.0.1", 8080);
client.setChannelHandler((sc)->{
@SneakyThrows ByteBuffer readBuffer = ByteBuffer.allocate(1024);
@Override //从channel读数据到缓冲区
protected void read(SocketChannel sc) { int readBytes = sc.read(readBuffer);
ByteBuffer readBuffer = ByteBuffer.allocate(1024); if (readBytes > 0) {
//从channel读数据到缓冲区 //Flips this buffer. The limit is set to the current position and then
int readBytes = sc.read(readBuffer); // the position is set to zero就是表示要从起始位置开始读取数据
if (readBytes > 0) { readBuffer.flip();
//Flips this buffer. The limit is set to the current position and then //returns the number of elements between the current position and the limit.
// the position is set to zero就是表示要从起始位置开始读取数据 // 要读取的字节长度
readBuffer.flip(); byte[] bytes = new byte[readBuffer.remaining()];
//returns the number of elements between the current position and the limit. //将缓冲区的数据读到bytes数组
// 要读取的字节长度 readBuffer.get(bytes);
byte[] bytes = new byte[readBuffer.remaining()]; String body = StrUtil.utf8Str(bytes);
//将缓冲区的数据读到bytes数组 Console.log("[{}]: {}", sc.getRemoteAddress(), body);
readBuffer.get(bytes); } else if (readBytes < 0) {
String body = StrUtil.utf8Str(bytes); sc.close();
Console.log("the read client receive message: " + body);
} else if (readBytes < 0) {
sc.close();
}
} }
}; });
client.listen(); client.listen();
ByteBuffer buffer = ByteBuffer.wrap("client 发生到 server".getBytes()); client.write(BufferUtil.createUtf8("你好。\n"));
client.write(buffer); client.write(BufferUtil.createUtf8("你好2。"));
buffer = ByteBuffer.wrap("client 再次发生到 server".getBytes());
client.write(buffer);
// 在控制台向服务器端发送数据 // 在控制台向服务器端发送数据
Console.log("在下方畅所欲言"); Console.log("输入发送的消息:");
Scanner scanner = new Scanner(System.in); Scanner scanner = new Scanner(System.in);
while (scanner.hasNextLine()) { while (scanner.hasNextLine()) {
String request = scanner.nextLine(); String request = scanner.nextLine();
if (request != null && request.trim().length() > 0) { if (request != null && request.trim().length() > 0) {
client.write( client.write(BufferUtil.createUtf8(request));
CharsetUtil.CHARSET_UTF_8
.encode("测试client" + ": " + request));
} }
} }
} }

View File

@ -1,5 +1,6 @@
package cn.hutool.socket.nio; package cn.hutool.socket.nio;
import cn.hutool.core.io.BufferUtil;
import cn.hutool.core.io.IORuntimeException; import cn.hutool.core.io.IORuntimeException;
import cn.hutool.core.io.IoUtil; import cn.hutool.core.io.IoUtil;
import cn.hutool.core.lang.Console; import cn.hutool.core.lang.Console;
@ -28,7 +29,8 @@ public class NioServerTest {
//将缓冲区的数据读到bytes数组 //将缓冲区的数据读到bytes数组
readBuffer.get(bytes); readBuffer.get(bytes);
String body = StrUtil.utf8Str(bytes); String body = StrUtil.utf8Str(bytes);
Console.log("the read server receive message: " + body); Console.log("[{}]: {}", sc.getRemoteAddress(), body);
doWrite(sc, body); doWrite(sc, body);
} else if (readBytes < 0) { } else if (readBytes < 0) {
IoUtil.close(sc); IoUtil.close(sc);
@ -41,16 +43,8 @@ public class NioServerTest {
} }
public static void doWrite(SocketChannel channel, String response) throws IOException { public static void doWrite(SocketChannel channel, String response) throws IOException {
response = "我们已收到消息:" + response; response = "收到消息:" + response;
if (!StrUtil.isBlank(response)) { //将缓冲数据写入渠道返回给客户端
byte[] bytes = response.getBytes(); channel.write(BufferUtil.createUtf8(response));
//分配一个bytes的length长度的ByteBuffer
ByteBuffer write = ByteBuffer.allocate(bytes.length);
//将返回数据写入缓冲区
write.put(bytes);
write.flip();
//将缓冲数据写入渠道返回给客户端
channel.write(write);
}
} }
} }