mirror of
https://gitee.com/chinabugotech/hutool.git
synced 2025-07-21 15:09:48 +08:00
change line sep
This commit is contained in:
Looly
@@ -1,118 +1,118 @@
|
||||
package cn.hutool.socket;
|
||||
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.core.util.RuntimeUtil;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* Socket通讯配置
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class SocketConfig implements Serializable{
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
/** CPU核心数 */
|
||||
private static final int CPU_COUNT = RuntimeUtil.getProcessorCount();
|
||||
|
||||
/** 共享线程池大小，此线程池用于接收和处理用户连接 */
|
||||
private int threadPoolSize = CPU_COUNT;
|
||||
|
||||
/** 读取超时时长，小于等于0表示默认 */
|
||||
private long readTimeout;
|
||||
/** 写出超时时长，小于等于0表示默认 */
|
||||
private long writeTimeout;
|
||||
|
||||
/** 读取缓存大小 */
|
||||
private int readBufferSize = IoUtil.DEFAULT_BUFFER_SIZE;
|
||||
/** 写出缓存大小 */
|
||||
private int writeBufferSize = IoUtil.DEFAULT_BUFFER_SIZE;
|
||||
|
||||
/**
|
||||
* 获取共享线程池大小，此线程池用于接收和处理用户连接
|
||||
*
|
||||
* @return 共享线程池大小，此线程池用于接收和处理用户连接
|
||||
*/
|
||||
public int getThreadPoolSize() {
|
||||
return threadPoolSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置共享线程池大小，此线程池用于接收和处理用户连接
|
||||
*
|
||||
* @param threadPoolSize 共享线程池大小，此线程池用于接收和处理用户连接
|
||||
*/
|
||||
public void setThreadPoolSize(int threadPoolSize) {
|
||||
this.threadPoolSize = threadPoolSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取读取超时时长，小于等于0表示默认
|
||||
*
|
||||
* @return 读取超时时长，小于等于0表示默认
|
||||
*/
|
||||
public long getReadTimeout() {
|
||||
return readTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置读取超时时长，小于等于0表示默认
|
||||
*
|
||||
* @param readTimeout 读取超时时长，小于等于0表示默认
|
||||
*/
|
||||
public void setReadTimeout(long readTimeout) {
|
||||
this.readTimeout = readTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取写出超时时长，小于等于0表示默认
|
||||
*
|
||||
* @return 写出超时时长，小于等于0表示默认
|
||||
*/
|
||||
public long getWriteTimeout() {
|
||||
return writeTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置写出超时时长，小于等于0表示默认
|
||||
*
|
||||
* @param writeTimeout 写出超时时长，小于等于0表示默认
|
||||
*/
|
||||
public void setWriteTimeout(long writeTimeout) {
|
||||
this.writeTimeout = writeTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取读取缓存大小
|
||||
* @return 读取缓存大小
|
||||
*/
|
||||
public int getReadBufferSize() {
|
||||
return readBufferSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置读取缓存大小
|
||||
* @param readBufferSize 读取缓存大小
|
||||
*/
|
||||
public void setReadBufferSize(int readBufferSize) {
|
||||
this.readBufferSize = readBufferSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取写出缓存大小
|
||||
* @return 写出缓存大小
|
||||
*/
|
||||
public int getWriteBufferSize() {
|
||||
return writeBufferSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置写出缓存大小
|
||||
* @param writeBufferSize 写出缓存大小
|
||||
*/
|
||||
public void setWriteBufferSize(int writeBufferSize) {
|
||||
this.writeBufferSize = writeBufferSize;
|
||||
}
|
||||
}
|
||||
package cn.hutool.socket;
|
||||
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.core.util.RuntimeUtil;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* Socket通讯配置
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class SocketConfig implements Serializable{
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
/** CPU核心数 */
|
||||
private static final int CPU_COUNT = RuntimeUtil.getProcessorCount();
|
||||
|
||||
/** 共享线程池大小，此线程池用于接收和处理用户连接 */
|
||||
private int threadPoolSize = CPU_COUNT;
|
||||
|
||||
/** 读取超时时长，小于等于0表示默认 */
|
||||
private long readTimeout;
|
||||
/** 写出超时时长，小于等于0表示默认 */
|
||||
private long writeTimeout;
|
||||
|
||||
/** 读取缓存大小 */
|
||||
private int readBufferSize = IoUtil.DEFAULT_BUFFER_SIZE;
|
||||
/** 写出缓存大小 */
|
||||
private int writeBufferSize = IoUtil.DEFAULT_BUFFER_SIZE;
|
||||
|
||||
/**
|
||||
* 获取共享线程池大小，此线程池用于接收和处理用户连接
|
||||
*
|
||||
* @return 共享线程池大小，此线程池用于接收和处理用户连接
|
||||
*/
|
||||
public int getThreadPoolSize() {
|
||||
return threadPoolSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置共享线程池大小，此线程池用于接收和处理用户连接
|
||||
*
|
||||
* @param threadPoolSize 共享线程池大小，此线程池用于接收和处理用户连接
|
||||
*/
|
||||
public void setThreadPoolSize(int threadPoolSize) {
|
||||
this.threadPoolSize = threadPoolSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取读取超时时长，小于等于0表示默认
|
||||
*
|
||||
* @return 读取超时时长，小于等于0表示默认
|
||||
*/
|
||||
public long getReadTimeout() {
|
||||
return readTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置读取超时时长，小于等于0表示默认
|
||||
*
|
||||
* @param readTimeout 读取超时时长，小于等于0表示默认
|
||||
*/
|
||||
public void setReadTimeout(long readTimeout) {
|
||||
this.readTimeout = readTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取写出超时时长，小于等于0表示默认
|
||||
*
|
||||
* @return 写出超时时长，小于等于0表示默认
|
||||
*/
|
||||
public long getWriteTimeout() {
|
||||
return writeTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置写出超时时长，小于等于0表示默认
|
||||
*
|
||||
* @param writeTimeout 写出超时时长，小于等于0表示默认
|
||||
*/
|
||||
public void setWriteTimeout(long writeTimeout) {
|
||||
this.writeTimeout = writeTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取读取缓存大小
|
||||
* @return 读取缓存大小
|
||||
*/
|
||||
public int getReadBufferSize() {
|
||||
return readBufferSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置读取缓存大小
|
||||
* @param readBufferSize 读取缓存大小
|
||||
*/
|
||||
public void setReadBufferSize(int readBufferSize) {
|
||||
this.readBufferSize = readBufferSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取写出缓存大小
|
||||
* @return 写出缓存大小
|
||||
*/
|
||||
public int getWriteBufferSize() {
|
||||
return writeBufferSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置写出缓存大小
|
||||
* @param writeBufferSize 写出缓存大小
|
||||
*/
|
||||
public void setWriteBufferSize(int writeBufferSize) {
|
||||
this.writeBufferSize = writeBufferSize;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,33 +1,33 @@
|
||||
package cn.hutool.socket;
|
||||
|
||||
import cn.hutool.core.exceptions.ExceptionUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
|
||||
/**
|
||||
* Socket异常
|
||||
*
|
||||
* @author xiaoleilu
|
||||
*/
|
||||
public class SocketRuntimeException extends RuntimeException {
|
||||
private static final long serialVersionUID = 8247610319171014183L;
|
||||
|
||||
public SocketRuntimeException(Throwable e) {
|
||||
super(ExceptionUtil.getMessage(e), e);
|
||||
}
|
||||
|
||||
public SocketRuntimeException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
public SocketRuntimeException(String messageTemplate, Object... params) {
|
||||
super(StrUtil.format(messageTemplate, params));
|
||||
}
|
||||
|
||||
public SocketRuntimeException(String message, Throwable throwable) {
|
||||
super(message, throwable);
|
||||
}
|
||||
|
||||
public SocketRuntimeException(Throwable throwable, String messageTemplate, Object... params) {
|
||||
super(StrUtil.format(messageTemplate, params), throwable);
|
||||
}
|
||||
}
|
||||
package cn.hutool.socket;
|
||||
|
||||
import cn.hutool.core.exceptions.ExceptionUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
|
||||
/**
|
||||
* Socket异常
|
||||
*
|
||||
* @author xiaoleilu
|
||||
*/
|
||||
public class SocketRuntimeException extends RuntimeException {
|
||||
private static final long serialVersionUID = 8247610319171014183L;
|
||||
|
||||
public SocketRuntimeException(Throwable e) {
|
||||
super(ExceptionUtil.getMessage(e), e);
|
||||
}
|
||||
|
||||
public SocketRuntimeException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
public SocketRuntimeException(String messageTemplate, Object... params) {
|
||||
super(StrUtil.format(messageTemplate, params));
|
||||
}
|
||||
|
||||
public SocketRuntimeException(String message, Throwable throwable) {
|
||||
super(message, throwable);
|
||||
}
|
||||
|
||||
public SocketRuntimeException(Throwable throwable, String messageTemplate, Object... params) {
|
||||
super(StrUtil.format(messageTemplate, params), throwable);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,46 +1,46 @@
|
||||
package cn.hutool.socket;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.SocketAddress;
|
||||
import java.nio.channels.AsynchronousSocketChannel;
|
||||
import java.nio.channels.ClosedChannelException;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
|
||||
/**
|
||||
* Socket相关工具类
|
||||
*
|
||||
* @author looly
|
||||
* @since 4.5.0
|
||||
*/
|
||||
public class SocketUtil {
|
||||
|
||||
/**
|
||||
* 获取远程端的地址信息，包括host和端口<br>
|
||||
* null表示channel为null或者远程主机未连接
|
||||
*
|
||||
* @param channel {@link AsynchronousSocketChannel}
|
||||
* @return 远程端的地址信息，包括host和端口，null表示channel为null或者远程主机未连接
|
||||
*/
|
||||
public static SocketAddress getRemoteAddress(AsynchronousSocketChannel channel) {
|
||||
try {
|
||||
return (null == channel) ? null : channel.getRemoteAddress();
|
||||
} catch (ClosedChannelException e) {
|
||||
// Channel未打开或已关闭，返回null表示未连接
|
||||
return null;
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 远程主机是否处于连接状态<br>
|
||||
* 通过判断远程地址获取成功与否判断
|
||||
*
|
||||
* @param channel {@link AsynchronousSocketChannel}
|
||||
* @return 远程主机是否处于连接状态
|
||||
*/
|
||||
public static boolean isConnected(AsynchronousSocketChannel channel) {
|
||||
return null != getRemoteAddress(channel);
|
||||
}
|
||||
}
|
||||
package cn.hutool.socket;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.SocketAddress;
|
||||
import java.nio.channels.AsynchronousSocketChannel;
|
||||
import java.nio.channels.ClosedChannelException;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
|
||||
/**
|
||||
* Socket相关工具类
|
||||
*
|
||||
* @author looly
|
||||
* @since 4.5.0
|
||||
*/
|
||||
public class SocketUtil {
|
||||
|
||||
/**
|
||||
* 获取远程端的地址信息，包括host和端口<br>
|
||||
* null表示channel为null或者远程主机未连接
|
||||
*
|
||||
* @param channel {@link AsynchronousSocketChannel}
|
||||
* @return 远程端的地址信息，包括host和端口，null表示channel为null或者远程主机未连接
|
||||
*/
|
||||
public static SocketAddress getRemoteAddress(AsynchronousSocketChannel channel) {
|
||||
try {
|
||||
return (null == channel) ? null : channel.getRemoteAddress();
|
||||
} catch (ClosedChannelException e) {
|
||||
// Channel未打开或已关闭，返回null表示未连接
|
||||
return null;
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 远程主机是否处于连接状态<br>
|
||||
* 通过判断远程地址获取成功与否判断
|
||||
*
|
||||
* @param channel {@link AsynchronousSocketChannel}
|
||||
* @return 远程主机是否处于连接状态
|
||||
*/
|
||||
public static boolean isConnected(AsynchronousSocketChannel channel) {
|
||||
return null != getRemoteAddress(channel);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,37 +1,37 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.AsynchronousSocketChannel;
|
||||
import java.nio.channels.CompletionHandler;
|
||||
|
||||
import cn.hutool.log.StaticLog;
|
||||
|
||||
/**
|
||||
* 接入完成回调，单例使用
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AioServer> {
|
||||
|
||||
@Override
|
||||
public void completed(AsynchronousSocketChannel socketChannel, AioServer aioServer) {
|
||||
// 继续等待接入（异步）
|
||||
aioServer.accept();
|
||||
|
||||
final IoAction<ByteBuffer> ioAction = aioServer.ioAction;
|
||||
// 创建Session会话
|
||||
final AioSession session = new AioSession(socketChannel, ioAction, aioServer.config);
|
||||
// 处理请求接入（同步）
|
||||
ioAction.accept(session);
|
||||
|
||||
// 处理读（异步）
|
||||
session.read();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void failed(Throwable exc, AioServer aioServer) {
|
||||
StaticLog.error(exc);
|
||||
}
|
||||
|
||||
}
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.AsynchronousSocketChannel;
|
||||
import java.nio.channels.CompletionHandler;
|
||||
|
||||
import cn.hutool.log.StaticLog;
|
||||
|
||||
/**
|
||||
* 接入完成回调，单例使用
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AioServer> {
|
||||
|
||||
@Override
|
||||
public void completed(AsynchronousSocketChannel socketChannel, AioServer aioServer) {
|
||||
// 继续等待接入（异步）
|
||||
aioServer.accept();
|
||||
|
||||
final IoAction<ByteBuffer> ioAction = aioServer.ioAction;
|
||||
// 创建Session会话
|
||||
final AioSession session = new AioSession(socketChannel, ioAction, aioServer.config);
|
||||
// 处理请求接入（同步）
|
||||
ioAction.accept(session);
|
||||
|
||||
// 处理读（异步）
|
||||
session.read();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void failed(Throwable exc, AioServer aioServer) {
|
||||
StaticLog.error(exc);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,142 +1,142 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketOption;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.AsynchronousChannelGroup;
|
||||
import java.nio.channels.AsynchronousSocketChannel;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.thread.ThreadFactoryBuilder;
|
||||
import cn.hutool.socket.SocketConfig;
|
||||
import cn.hutool.socket.SocketRuntimeException;
|
||||
|
||||
/**
|
||||
* Aio Socket客户端
|
||||
*
|
||||
* @author looly
|
||||
* @since 4.5.0
|
||||
*/
|
||||
public class AioClient implements Closeable{
|
||||
|
||||
private final AioSession session;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param address 地址
|
||||
* @param ioAction IO处理类
|
||||
*/
|
||||
public AioClient(InetSocketAddress address, IoAction<ByteBuffer> ioAction) {
|
||||
this(address, ioAction, new SocketConfig());
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param address 地址
|
||||
* @param ioAction IO处理类
|
||||
* @param config 配置项
|
||||
*/
|
||||
public AioClient(InetSocketAddress address, IoAction<ByteBuffer> ioAction, SocketConfig config) {
|
||||
this(createChannel(address, config.getThreadPoolSize()), ioAction, config);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param channel {@link AsynchronousSocketChannel}
|
||||
* @param ioAction IO处理类
|
||||
* @param config 配置项
|
||||
*/
|
||||
public AioClient(AsynchronousSocketChannel channel, IoAction<ByteBuffer> ioAction, SocketConfig config) {
|
||||
this.session = new AioSession(channel, ioAction, config);
|
||||
ioAction.accept(this.session);
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置 Socket 的 Option 选项<br>
|
||||
* 选项见：{@link java.net.StandardSocketOptions}
|
||||
*
|
||||
* @param <T> 选项泛型
|
||||
* @param name {@link SocketOption} 枚举
|
||||
* @param value SocketOption参数
|
||||
* @return this
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
public <T> AioClient setOption(SocketOption<T> name, T value) throws IOException {
|
||||
this.session.getChannel().setOption(name, value);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取IO处理器
|
||||
*
|
||||
* @return {@link IoAction}
|
||||
*/
|
||||
public IoAction<ByteBuffer> getIoAction() {
|
||||
return this.session.getIoAction();
|
||||
}
|
||||
|
||||
/**
|
||||
* 从服务端读取数据
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioClient read() {
|
||||
this.session.read();
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到服务端
|
||||
*
|
||||
* @param data 数据
|
||||
* @return this
|
||||
*/
|
||||
public AioClient write(ByteBuffer data) {
|
||||
this.session.write(data);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭客户端
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
this.session.close();
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------------------- Private method start
|
||||
/**
|
||||
* 初始化
|
||||
*
|
||||
* @param address 地址和端口
|
||||
* @param poolSize 线程池大小
|
||||
* @return this
|
||||
*/
|
||||
private static AsynchronousSocketChannel createChannel(InetSocketAddress address, int poolSize) {
|
||||
|
||||
AsynchronousSocketChannel channel;
|
||||
try {
|
||||
AsynchronousChannelGroup group = AsynchronousChannelGroup.withFixedThreadPool(//
|
||||
poolSize, // 默认线程池大小
|
||||
ThreadFactoryBuilder.create().setNamePrefix("Huool-socket-").build()//
|
||||
);
|
||||
channel = AsynchronousSocketChannel.open(group);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
|
||||
try {
|
||||
channel.connect(address).get();
|
||||
} catch (InterruptedException | ExecutionException e) {
|
||||
throw new SocketRuntimeException(e);
|
||||
}
|
||||
return channel;
|
||||
}
|
||||
// ------------------------------------------------------------------------------------- Private method end
|
||||
}
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketOption;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.AsynchronousChannelGroup;
|
||||
import java.nio.channels.AsynchronousSocketChannel;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.thread.ThreadFactoryBuilder;
|
||||
import cn.hutool.socket.SocketConfig;
|
||||
import cn.hutool.socket.SocketRuntimeException;
|
||||
|
||||
/**
|
||||
* Aio Socket客户端
|
||||
*
|
||||
* @author looly
|
||||
* @since 4.5.0
|
||||
*/
|
||||
public class AioClient implements Closeable{
|
||||
|
||||
private final AioSession session;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param address 地址
|
||||
* @param ioAction IO处理类
|
||||
*/
|
||||
public AioClient(InetSocketAddress address, IoAction<ByteBuffer> ioAction) {
|
||||
this(address, ioAction, new SocketConfig());
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param address 地址
|
||||
* @param ioAction IO处理类
|
||||
* @param config 配置项
|
||||
*/
|
||||
public AioClient(InetSocketAddress address, IoAction<ByteBuffer> ioAction, SocketConfig config) {
|
||||
this(createChannel(address, config.getThreadPoolSize()), ioAction, config);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param channel {@link AsynchronousSocketChannel}
|
||||
* @param ioAction IO处理类
|
||||
* @param config 配置项
|
||||
*/
|
||||
public AioClient(AsynchronousSocketChannel channel, IoAction<ByteBuffer> ioAction, SocketConfig config) {
|
||||
this.session = new AioSession(channel, ioAction, config);
|
||||
ioAction.accept(this.session);
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置 Socket 的 Option 选项<br>
|
||||
* 选项见：{@link java.net.StandardSocketOptions}
|
||||
*
|
||||
* @param <T> 选项泛型
|
||||
* @param name {@link SocketOption} 枚举
|
||||
* @param value SocketOption参数
|
||||
* @return this
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
public <T> AioClient setOption(SocketOption<T> name, T value) throws IOException {
|
||||
this.session.getChannel().setOption(name, value);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取IO处理器
|
||||
*
|
||||
* @return {@link IoAction}
|
||||
*/
|
||||
public IoAction<ByteBuffer> getIoAction() {
|
||||
return this.session.getIoAction();
|
||||
}
|
||||
|
||||
/**
|
||||
* 从服务端读取数据
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioClient read() {
|
||||
this.session.read();
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到服务端
|
||||
*
|
||||
* @param data 数据
|
||||
* @return this
|
||||
*/
|
||||
public AioClient write(ByteBuffer data) {
|
||||
this.session.write(data);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭客户端
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
this.session.close();
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------------------- Private method start
|
||||
/**
|
||||
* 初始化
|
||||
*
|
||||
* @param address 地址和端口
|
||||
* @param poolSize 线程池大小
|
||||
* @return this
|
||||
*/
|
||||
private static AsynchronousSocketChannel createChannel(InetSocketAddress address, int poolSize) {
|
||||
|
||||
AsynchronousSocketChannel channel;
|
||||
try {
|
||||
AsynchronousChannelGroup group = AsynchronousChannelGroup.withFixedThreadPool(//
|
||||
poolSize, // 默认线程池大小
|
||||
ThreadFactoryBuilder.create().setNamePrefix("Huool-socket-").build()//
|
||||
);
|
||||
channel = AsynchronousSocketChannel.open(group);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
|
||||
try {
|
||||
channel.connect(address).get();
|
||||
} catch (InterruptedException | ExecutionException e) {
|
||||
throw new SocketRuntimeException(e);
|
||||
}
|
||||
return channel;
|
||||
}
|
||||
// ------------------------------------------------------------------------------------- Private method end
|
||||
}
|
||||
|
||||
@@ -1,184 +1,184 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.core.thread.ThreadFactoryBuilder;
|
||||
import cn.hutool.core.thread.ThreadUtil;
|
||||
import cn.hutool.log.Log;
|
||||
import cn.hutool.log.LogFactory;
|
||||
import cn.hutool.socket.SocketConfig;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketOption;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.AsynchronousChannelGroup;
|
||||
import java.nio.channels.AsynchronousServerSocketChannel;
|
||||
|
||||
/**
|
||||
* 基于AIO的Socket服务端实现
|
||||
*
|
||||
* @author looly
|
||||
*/
|
||||
public class AioServer implements Closeable {
|
||||
private static final Log log = LogFactory.get();
|
||||
private static final AcceptHandler ACCEPT_HANDLER = new AcceptHandler();
|
||||
|
||||
private AsynchronousChannelGroup group;
|
||||
private AsynchronousServerSocketChannel channel;
|
||||
protected IoAction<ByteBuffer> ioAction;
|
||||
protected final SocketConfig config;
|
||||
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param port 端口
|
||||
*/
|
||||
public AioServer(int port) {
|
||||
this(new InetSocketAddress(port), new SocketConfig());
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param address 地址
|
||||
* @param config {@link SocketConfig} 配置项
|
||||
*/
|
||||
public AioServer(InetSocketAddress address, SocketConfig config) {
|
||||
this.config = config;
|
||||
init(address);
|
||||
}
|
||||
|
||||
/**
|
||||
* 初始化
|
||||
*
|
||||
* @param address 地址和端口
|
||||
* @return this
|
||||
*/
|
||||
public AioServer init(InetSocketAddress address) {
|
||||
try {
|
||||
this.group = AsynchronousChannelGroup.withFixedThreadPool(//
|
||||
config.getThreadPoolSize(), // 默认线程池大小
|
||||
ThreadFactoryBuilder.create().setNamePrefix("Hutool-socket-").build()//
|
||||
);
|
||||
this.channel = AsynchronousServerSocketChannel.open(group).bind(address);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @param sync 是否阻塞
|
||||
*/
|
||||
public void start(boolean sync) {
|
||||
doStart(sync);
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置 Socket 的 Option 选项<br>
|
||||
* 选项见：{@link java.net.StandardSocketOptions}
|
||||
*
|
||||
* @param <T> 选项泛型
|
||||
* @param name {@link SocketOption} 枚举
|
||||
* @param value SocketOption参数
|
||||
* @return this
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
public <T> AioServer setOption(SocketOption<T> name, T value) throws IOException {
|
||||
this.channel.setOption(name, value);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取IO处理器
|
||||
*
|
||||
* @return {@link IoAction}
|
||||
*/
|
||||
public IoAction<ByteBuffer> getIoAction() {
|
||||
return this.ioAction;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置IO处理器，单例存在
|
||||
*
|
||||
* @param ioAction {@link IoAction}
|
||||
* @return this;
|
||||
*/
|
||||
public AioServer setIoAction(IoAction<ByteBuffer> ioAction) {
|
||||
this.ioAction = ioAction;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取{@link AsynchronousServerSocketChannel}
|
||||
*
|
||||
* @return {@link AsynchronousServerSocketChannel}
|
||||
*/
|
||||
public AsynchronousServerSocketChannel getChannel() {
|
||||
return this.channel;
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理接入的客户端
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioServer accept() {
|
||||
this.channel.accept(this, ACCEPT_HANDLER);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 服务是否开启状态
|
||||
*
|
||||
* @return 服务是否开启状态
|
||||
*/
|
||||
public boolean isOpen() {
|
||||
return (null != this.channel) && this.channel.isOpen();
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭服务
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
IoUtil.close(this.channel);
|
||||
|
||||
if (null != this.group && false == this.group.isShutdown()) {
|
||||
try {
|
||||
this.group.shutdownNow();
|
||||
} catch (IOException e) {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
|
||||
// 结束阻塞
|
||||
synchronized (this) {
|
||||
this.notify();
|
||||
}
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------------------- Private method start
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @param sync 是否阻塞
|
||||
*/
|
||||
private void doStart(boolean sync) {
|
||||
log.debug("Aio Server started, waiting for accept.");
|
||||
|
||||
// 接收客户端连接
|
||||
accept();
|
||||
|
||||
if (sync) {
|
||||
ThreadUtil.sync(this);
|
||||
}
|
||||
}
|
||||
// ------------------------------------------------------------------------------------- Private method end
|
||||
}
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.core.thread.ThreadFactoryBuilder;
|
||||
import cn.hutool.core.thread.ThreadUtil;
|
||||
import cn.hutool.log.Log;
|
||||
import cn.hutool.log.LogFactory;
|
||||
import cn.hutool.socket.SocketConfig;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketOption;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.AsynchronousChannelGroup;
|
||||
import java.nio.channels.AsynchronousServerSocketChannel;
|
||||
|
||||
/**
|
||||
* 基于AIO的Socket服务端实现
|
||||
*
|
||||
* @author looly
|
||||
*/
|
||||
public class AioServer implements Closeable {
|
||||
private static final Log log = LogFactory.get();
|
||||
private static final AcceptHandler ACCEPT_HANDLER = new AcceptHandler();
|
||||
|
||||
private AsynchronousChannelGroup group;
|
||||
private AsynchronousServerSocketChannel channel;
|
||||
protected IoAction<ByteBuffer> ioAction;
|
||||
protected final SocketConfig config;
|
||||
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param port 端口
|
||||
*/
|
||||
public AioServer(int port) {
|
||||
this(new InetSocketAddress(port), new SocketConfig());
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param address 地址
|
||||
* @param config {@link SocketConfig} 配置项
|
||||
*/
|
||||
public AioServer(InetSocketAddress address, SocketConfig config) {
|
||||
this.config = config;
|
||||
init(address);
|
||||
}
|
||||
|
||||
/**
|
||||
* 初始化
|
||||
*
|
||||
* @param address 地址和端口
|
||||
* @return this
|
||||
*/
|
||||
public AioServer init(InetSocketAddress address) {
|
||||
try {
|
||||
this.group = AsynchronousChannelGroup.withFixedThreadPool(//
|
||||
config.getThreadPoolSize(), // 默认线程池大小
|
||||
ThreadFactoryBuilder.create().setNamePrefix("Hutool-socket-").build()//
|
||||
);
|
||||
this.channel = AsynchronousServerSocketChannel.open(group).bind(address);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @param sync 是否阻塞
|
||||
*/
|
||||
public void start(boolean sync) {
|
||||
doStart(sync);
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置 Socket 的 Option 选项<br>
|
||||
* 选项见：{@link java.net.StandardSocketOptions}
|
||||
*
|
||||
* @param <T> 选项泛型
|
||||
* @param name {@link SocketOption} 枚举
|
||||
* @param value SocketOption参数
|
||||
* @return this
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
public <T> AioServer setOption(SocketOption<T> name, T value) throws IOException {
|
||||
this.channel.setOption(name, value);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取IO处理器
|
||||
*
|
||||
* @return {@link IoAction}
|
||||
*/
|
||||
public IoAction<ByteBuffer> getIoAction() {
|
||||
return this.ioAction;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置IO处理器，单例存在
|
||||
*
|
||||
* @param ioAction {@link IoAction}
|
||||
* @return this;
|
||||
*/
|
||||
public AioServer setIoAction(IoAction<ByteBuffer> ioAction) {
|
||||
this.ioAction = ioAction;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取{@link AsynchronousServerSocketChannel}
|
||||
*
|
||||
* @return {@link AsynchronousServerSocketChannel}
|
||||
*/
|
||||
public AsynchronousServerSocketChannel getChannel() {
|
||||
return this.channel;
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理接入的客户端
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioServer accept() {
|
||||
this.channel.accept(this, ACCEPT_HANDLER);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 服务是否开启状态
|
||||
*
|
||||
* @return 服务是否开启状态
|
||||
*/
|
||||
public boolean isOpen() {
|
||||
return (null != this.channel) && this.channel.isOpen();
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭服务
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
IoUtil.close(this.channel);
|
||||
|
||||
if (null != this.group && false == this.group.isShutdown()) {
|
||||
try {
|
||||
this.group.shutdownNow();
|
||||
} catch (IOException e) {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
|
||||
// 结束阻塞
|
||||
synchronized (this) {
|
||||
this.notify();
|
||||
}
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------------------- Private method start
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @param sync 是否阻塞
|
||||
*/
|
||||
private void doStart(boolean sync) {
|
||||
log.debug("Aio Server started, waiting for accept.");
|
||||
|
||||
// 接收客户端连接
|
||||
accept();
|
||||
|
||||
if (sync) {
|
||||
ThreadUtil.sync(this);
|
||||
}
|
||||
}
|
||||
// ------------------------------------------------------------------------------------- Private method end
|
||||
}
|
||||
|
||||
@@ -1,214 +1,214 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.socket.SocketConfig;
|
||||
import cn.hutool.socket.SocketUtil;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.SocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.AsynchronousSocketChannel;
|
||||
import java.nio.channels.CompletionHandler;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* AIO会话<br>
|
||||
* 每个客户端对应一个会话对象
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class AioSession implements Closeable{
|
||||
|
||||
private static final ReadHandler READ_HANDLER = new ReadHandler();
|
||||
|
||||
private final AsynchronousSocketChannel channel;
|
||||
private final IoAction<ByteBuffer> ioAction;
|
||||
private ByteBuffer readBuffer;
|
||||
private ByteBuffer writeBuffer;
|
||||
/** 读取超时时长，小于等于0表示默认 */
|
||||
private final long readTimeout;
|
||||
/** 写出超时时长，小于等于0表示默认 */
|
||||
private final long writeTimeout;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param channel {@link AsynchronousSocketChannel}
|
||||
* @param ioAction IO消息处理类
|
||||
* @param config 配置项
|
||||
*/
|
||||
public AioSession(AsynchronousSocketChannel channel, IoAction<ByteBuffer> ioAction, SocketConfig config) {
|
||||
this.channel = channel;
|
||||
this.ioAction = ioAction;
|
||||
|
||||
this.readBuffer = ByteBuffer.allocate(config.getReadBufferSize());
|
||||
this.writeBuffer = ByteBuffer.allocate(config.getWriteBufferSize());
|
||||
this.readTimeout = config.getReadTimeout();
|
||||
this.writeTimeout = config.getWriteTimeout();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取{@link AsynchronousSocketChannel}
|
||||
*
|
||||
* @return {@link AsynchronousSocketChannel}
|
||||
*/
|
||||
public AsynchronousSocketChannel getChannel() {
|
||||
return this.channel;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取读取Buffer
|
||||
*
|
||||
* @return 读取Buffer
|
||||
*/
|
||||
public ByteBuffer getReadBuffer() {
|
||||
return this.readBuffer;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取写Buffer
|
||||
*
|
||||
* @return 写Buffer
|
||||
*/
|
||||
public ByteBuffer getWriteBuffer() {
|
||||
return this.writeBuffer;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取消息处理器
|
||||
*
|
||||
* @return {@link IoAction}
|
||||
*/
|
||||
public IoAction<ByteBuffer> getIoAction() {
|
||||
return this.ioAction;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取远程主机（客户端）地址和端口
|
||||
*
|
||||
* @return 远程主机（客户端）地址和端口
|
||||
*/
|
||||
public SocketAddress getRemoteAddress() {
|
||||
return SocketUtil.getRemoteAddress(this.channel);
|
||||
}
|
||||
|
||||
/**
|
||||
* 读取数据到Buffer
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioSession read() {
|
||||
return read(READ_HANDLER);
|
||||
}
|
||||
|
||||
/**
|
||||
* 读取数据到Buffer
|
||||
*
|
||||
* @param handler {@link CompletionHandler}
|
||||
* @return this
|
||||
*/
|
||||
public AioSession read(CompletionHandler<Integer, AioSession> handler) {
|
||||
if (isOpen()) {
|
||||
this.readBuffer.clear();
|
||||
this.channel.read(this.readBuffer, Math.max(this.readTimeout, 0L), TimeUnit.MILLISECONDS, this, handler);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到目标端，并关闭输出
|
||||
*
|
||||
* @param data 数据
|
||||
* @return this
|
||||
*/
|
||||
public AioSession writeAndClose(ByteBuffer data) {
|
||||
write(data);
|
||||
return closeOut();
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到目标端
|
||||
*
|
||||
* @param data 数据
|
||||
* @return {@link Future}
|
||||
*/
|
||||
public Future<Integer> write(ByteBuffer data) {
|
||||
return this.channel.write(data);
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到目标端
|
||||
*
|
||||
* @param data 数据
|
||||
* @param handler {@link CompletionHandler}
|
||||
* @return this
|
||||
*/
|
||||
public AioSession write(ByteBuffer data, CompletionHandler<Integer, AioSession> handler) {
|
||||
this.channel.write(data, Math.max(this.writeTimeout, 0L), TimeUnit.MILLISECONDS, this, handler);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 会话是否打开状态<br>
|
||||
* 当Socket保持连接时会话始终打开
|
||||
*
|
||||
* @return 会话是否打开状态
|
||||
*/
|
||||
public boolean isOpen() {
|
||||
return (null != this.channel) && this.channel.isOpen();
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭输出
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioSession closeIn() {
|
||||
if (null != this.channel) {
|
||||
try {
|
||||
this.channel.shutdownInput();
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭输出
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioSession closeOut() {
|
||||
if (null != this.channel) {
|
||||
try {
|
||||
this.channel.shutdownOutput();
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭会话
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
IoUtil.close(this.channel);
|
||||
this.readBuffer = null;
|
||||
this.writeBuffer = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行读，用于读取事件结束的回调
|
||||
*/
|
||||
protected void callbackRead() {
|
||||
readBuffer.flip();// 读模式
|
||||
ioAction.doAction(this, readBuffer);
|
||||
}
|
||||
}
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.socket.SocketConfig;
|
||||
import cn.hutool.socket.SocketUtil;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.SocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.AsynchronousSocketChannel;
|
||||
import java.nio.channels.CompletionHandler;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* AIO会话<br>
|
||||
* 每个客户端对应一个会话对象
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class AioSession implements Closeable{
|
||||
|
||||
private static final ReadHandler READ_HANDLER = new ReadHandler();
|
||||
|
||||
private final AsynchronousSocketChannel channel;
|
||||
private final IoAction<ByteBuffer> ioAction;
|
||||
private ByteBuffer readBuffer;
|
||||
private ByteBuffer writeBuffer;
|
||||
/** 读取超时时长，小于等于0表示默认 */
|
||||
private final long readTimeout;
|
||||
/** 写出超时时长，小于等于0表示默认 */
|
||||
private final long writeTimeout;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param channel {@link AsynchronousSocketChannel}
|
||||
* @param ioAction IO消息处理类
|
||||
* @param config 配置项
|
||||
*/
|
||||
public AioSession(AsynchronousSocketChannel channel, IoAction<ByteBuffer> ioAction, SocketConfig config) {
|
||||
this.channel = channel;
|
||||
this.ioAction = ioAction;
|
||||
|
||||
this.readBuffer = ByteBuffer.allocate(config.getReadBufferSize());
|
||||
this.writeBuffer = ByteBuffer.allocate(config.getWriteBufferSize());
|
||||
this.readTimeout = config.getReadTimeout();
|
||||
this.writeTimeout = config.getWriteTimeout();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取{@link AsynchronousSocketChannel}
|
||||
*
|
||||
* @return {@link AsynchronousSocketChannel}
|
||||
*/
|
||||
public AsynchronousSocketChannel getChannel() {
|
||||
return this.channel;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取读取Buffer
|
||||
*
|
||||
* @return 读取Buffer
|
||||
*/
|
||||
public ByteBuffer getReadBuffer() {
|
||||
return this.readBuffer;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取写Buffer
|
||||
*
|
||||
* @return 写Buffer
|
||||
*/
|
||||
public ByteBuffer getWriteBuffer() {
|
||||
return this.writeBuffer;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取消息处理器
|
||||
*
|
||||
* @return {@link IoAction}
|
||||
*/
|
||||
public IoAction<ByteBuffer> getIoAction() {
|
||||
return this.ioAction;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取远程主机（客户端）地址和端口
|
||||
*
|
||||
* @return 远程主机（客户端）地址和端口
|
||||
*/
|
||||
public SocketAddress getRemoteAddress() {
|
||||
return SocketUtil.getRemoteAddress(this.channel);
|
||||
}
|
||||
|
||||
/**
|
||||
* 读取数据到Buffer
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioSession read() {
|
||||
return read(READ_HANDLER);
|
||||
}
|
||||
|
||||
/**
|
||||
* 读取数据到Buffer
|
||||
*
|
||||
* @param handler {@link CompletionHandler}
|
||||
* @return this
|
||||
*/
|
||||
public AioSession read(CompletionHandler<Integer, AioSession> handler) {
|
||||
if (isOpen()) {
|
||||
this.readBuffer.clear();
|
||||
this.channel.read(this.readBuffer, Math.max(this.readTimeout, 0L), TimeUnit.MILLISECONDS, this, handler);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到目标端，并关闭输出
|
||||
*
|
||||
* @param data 数据
|
||||
* @return this
|
||||
*/
|
||||
public AioSession writeAndClose(ByteBuffer data) {
|
||||
write(data);
|
||||
return closeOut();
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到目标端
|
||||
*
|
||||
* @param data 数据
|
||||
* @return {@link Future}
|
||||
*/
|
||||
public Future<Integer> write(ByteBuffer data) {
|
||||
return this.channel.write(data);
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到目标端
|
||||
*
|
||||
* @param data 数据
|
||||
* @param handler {@link CompletionHandler}
|
||||
* @return this
|
||||
*/
|
||||
public AioSession write(ByteBuffer data, CompletionHandler<Integer, AioSession> handler) {
|
||||
this.channel.write(data, Math.max(this.writeTimeout, 0L), TimeUnit.MILLISECONDS, this, handler);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 会话是否打开状态<br>
|
||||
* 当Socket保持连接时会话始终打开
|
||||
*
|
||||
* @return 会话是否打开状态
|
||||
*/
|
||||
public boolean isOpen() {
|
||||
return (null != this.channel) && this.channel.isOpen();
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭输出
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioSession closeIn() {
|
||||
if (null != this.channel) {
|
||||
try {
|
||||
this.channel.shutdownInput();
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭输出
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioSession closeOut() {
|
||||
if (null != this.channel) {
|
||||
try {
|
||||
this.channel.shutdownOutput();
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭会话
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
IoUtil.close(this.channel);
|
||||
this.readBuffer = null;
|
||||
this.writeBuffer = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行读，用于读取事件结束的回调
|
||||
*/
|
||||
protected void callbackRead() {
|
||||
readBuffer.flip();// 读模式
|
||||
ioAction.doAction(this, readBuffer);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,35 +1,35 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
/**
|
||||
* Socket流处理接口<br>
|
||||
* 实现此接口用于处理接收到的消息，发送指定消息
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
* @param <T> 经过解码器解码后的数据类型
|
||||
*/
|
||||
public interface IoAction<T> {
|
||||
|
||||
/**
|
||||
* 接收客户端连接（会话建立）事件处理
|
||||
*
|
||||
* @param session 会话
|
||||
*/
|
||||
void accept(AioSession session);
|
||||
|
||||
/**
|
||||
* 执行数据处理（消息读取）
|
||||
*
|
||||
* @param session Socket Session会话
|
||||
* @param data 解码后的数据
|
||||
*/
|
||||
void doAction(AioSession session, T data);
|
||||
|
||||
/**
|
||||
* 数据读取失败的回调事件处理（消息读取失败）
|
||||
*
|
||||
* @param exc 异常
|
||||
* @param session Session
|
||||
*/
|
||||
void failed(Throwable exc, AioSession session);
|
||||
}
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
/**
|
||||
* Socket流处理接口<br>
|
||||
* 实现此接口用于处理接收到的消息，发送指定消息
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
* @param <T> 经过解码器解码后的数据类型
|
||||
*/
|
||||
public interface IoAction<T> {
|
||||
|
||||
/**
|
||||
* 接收客户端连接（会话建立）事件处理
|
||||
*
|
||||
* @param session 会话
|
||||
*/
|
||||
void accept(AioSession session);
|
||||
|
||||
/**
|
||||
* 执行数据处理（消息读取）
|
||||
*
|
||||
* @param session Socket Session会话
|
||||
* @param data 解码后的数据
|
||||
*/
|
||||
void doAction(AioSession session, T data);
|
||||
|
||||
/**
|
||||
* 数据读取失败的回调事件处理（消息读取失败）
|
||||
*
|
||||
* @param exc 异常
|
||||
* @param session Session
|
||||
*/
|
||||
void failed(Throwable exc, AioSession session);
|
||||
}
|
||||
|
||||
@@ -1,25 +1,25 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import java.nio.channels.CompletionHandler;
|
||||
|
||||
import cn.hutool.socket.SocketRuntimeException;
|
||||
|
||||
/**
|
||||
* 数据读取完成回调，调用Session中相应方法处理消息，单例使用
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class ReadHandler implements CompletionHandler<Integer, AioSession> {
|
||||
|
||||
@Override
|
||||
public void completed(Integer result, AioSession session) {
|
||||
session.callbackRead();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void failed(Throwable exc, AioSession session) {
|
||||
throw new SocketRuntimeException(exc);
|
||||
}
|
||||
|
||||
}
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import java.nio.channels.CompletionHandler;
|
||||
|
||||
import cn.hutool.socket.SocketRuntimeException;
|
||||
|
||||
/**
|
||||
* 数据读取完成回调，调用Session中相应方法处理消息，单例使用
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class ReadHandler implements CompletionHandler<Integer, AioSession> {
|
||||
|
||||
@Override
|
||||
public void completed(Integer result, AioSession session) {
|
||||
session.callbackRead();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void failed(Throwable exc, AioSession session) {
|
||||
throw new SocketRuntimeException(exc);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,24 +1,24 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import cn.hutool.log.StaticLog;
|
||||
|
||||
/**
|
||||
* 简易IO信息处理类<br>
|
||||
* 简单实现了accept和failed事件
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public abstract class SimpleIoAction implements IoAction<ByteBuffer> {
|
||||
|
||||
@Override
|
||||
public void accept(AioSession session) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void failed(Throwable exc, AioSession session) {
|
||||
StaticLog.error(exc);
|
||||
}
|
||||
}
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import cn.hutool.log.StaticLog;
|
||||
|
||||
/**
|
||||
* 简易IO信息处理类<br>
|
||||
* 简单实现了accept和failed事件
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public abstract class SimpleIoAction implements IoAction<ByteBuffer> {
|
||||
|
||||
@Override
|
||||
public void accept(AioSession session) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void failed(Throwable exc, AioSession session) {
|
||||
StaticLog.error(exc);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
/**
|
||||
* AIO相关封装
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
/**
|
||||
* AIO相关封装
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
package cn.hutool.socket.aio;
|
||||
@@ -1,38 +1,38 @@
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.log.StaticLog;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.CompletionHandler;
|
||||
import java.nio.channels.ServerSocketChannel;
|
||||
import java.nio.channels.SocketChannel;
|
||||
|
||||
/**
|
||||
* 接入完成回调，单例使用
|
||||
*
|
||||
* @author looly
|
||||
*/
|
||||
public class AcceptHandler implements CompletionHandler<ServerSocketChannel, NioServer> {
|
||||
|
||||
@Override
|
||||
public void completed(ServerSocketChannel serverSocketChannel, NioServer nioServer) {
|
||||
SocketChannel socketChannel;
|
||||
try {
|
||||
// 获取连接到此服务器的客户端通道
|
||||
socketChannel = serverSocketChannel.accept();
|
||||
StaticLog.debug("Client [{}] accepted.", socketChannel.getRemoteAddress());
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
|
||||
// SocketChannel通道的可读事件注册到Selector中
|
||||
NioUtil.registerChannel(nioServer.getSelector(), socketChannel, Operation.READ);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void failed(Throwable exc, NioServer nioServer) {
|
||||
StaticLog.error(exc);
|
||||
}
|
||||
|
||||
}
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.log.StaticLog;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.CompletionHandler;
|
||||
import java.nio.channels.ServerSocketChannel;
|
||||
import java.nio.channels.SocketChannel;
|
||||
|
||||
/**
|
||||
* 接入完成回调，单例使用
|
||||
*
|
||||
* @author looly
|
||||
*/
|
||||
public class AcceptHandler implements CompletionHandler<ServerSocketChannel, NioServer> {
|
||||
|
||||
@Override
|
||||
public void completed(ServerSocketChannel serverSocketChannel, NioServer nioServer) {
|
||||
SocketChannel socketChannel;
|
||||
try {
|
||||
// 获取连接到此服务器的客户端通道
|
||||
socketChannel = serverSocketChannel.accept();
|
||||
StaticLog.debug("Client [{}] accepted.", socketChannel.getRemoteAddress());
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
|
||||
// SocketChannel通道的可读事件注册到Selector中
|
||||
NioUtil.registerChannel(nioServer.getSelector(), socketChannel, Operation.READ);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void failed(Throwable exc, NioServer nioServer) {
|
||||
StaticLog.error(exc);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,19 +1,19 @@
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import java.nio.channels.SocketChannel;
|
||||
|
||||
/**
|
||||
* NIO数据处理接口，通过实现此接口，可以从{@link SocketChannel}中读写数据
|
||||
*
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface ChannelHandler {
|
||||
|
||||
/**
|
||||
* 处理NIO数据
|
||||
*
|
||||
* @param socketChannel {@link SocketChannel}
|
||||
* @throws Exception 可能的处理异常
|
||||
*/
|
||||
void handle(SocketChannel socketChannel) throws Exception;
|
||||
}
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import java.nio.channels.SocketChannel;
|
||||
|
||||
/**
|
||||
* NIO数据处理接口，通过实现此接口，可以从{@link SocketChannel}中读写数据
|
||||
*
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface ChannelHandler {
|
||||
|
||||
/**
|
||||
* 处理NIO数据
|
||||
*
|
||||
* @param socketChannel {@link SocketChannel}
|
||||
* @throws Exception 可能的处理异常
|
||||
*/
|
||||
void handle(SocketChannel socketChannel) throws Exception;
|
||||
}
|
||||
|
||||
@@ -1,162 +1,162 @@
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.core.thread.ThreadUtil;
|
||||
import cn.hutool.socket.SocketRuntimeException;
|
||||
|
||||
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.SocketChannel;
|
||||
import java.util.Iterator;
|
||||
|
||||
/**
|
||||
* NIO客户端
|
||||
*
|
||||
* @author looly
|
||||
* @since 4.4.5
|
||||
*/
|
||||
public class NioClient implements Closeable {
|
||||
|
||||
private Selector selector;
|
||||
private SocketChannel channel;
|
||||
private ChannelHandler handler;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param host 服务器地址
|
||||
* @param port 端口
|
||||
*/
|
||||
public NioClient(String host, int port) {
|
||||
init(new InetSocketAddress(host, port));
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param address 服务器地址
|
||||
*/
|
||||
public NioClient(InetSocketAddress address) {
|
||||
init(address);
|
||||
}
|
||||
|
||||
/**
|
||||
* 初始化
|
||||
*
|
||||
* @param address 地址和端口
|
||||
* @return this
|
||||
*/
|
||||
public NioClient init(InetSocketAddress address) {
|
||||
try {
|
||||
//创建一个SocketChannel对象，配置成非阻塞模式
|
||||
this.channel = SocketChannel.open();
|
||||
channel.configureBlocking(false);
|
||||
channel.connect(address);
|
||||
|
||||
//创建一个选择器，并把SocketChannel交给selector对象
|
||||
this.selector = Selector.open();
|
||||
channel.register(this.selector, SelectionKey.OP_READ);
|
||||
|
||||
// 等待建立连接
|
||||
//noinspection StatementWithEmptyBody
|
||||
while (false == channel.finishConnect()){}
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置NIO数据处理器
|
||||
*
|
||||
* @param handler {@link ChannelHandler}
|
||||
* @return this
|
||||
*/
|
||||
public NioClient setChannelHandler(ChannelHandler handler){
|
||||
this.handler = handler;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*/
|
||||
public void listen() {
|
||||
ThreadUtil.execute(() -> {
|
||||
try {
|
||||
doListen();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
private void doListen() throws IOException {
|
||||
while (this.selector.isOpen() && 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) {
|
||||
// 读事件就绪
|
||||
if (key.isReadable()) {
|
||||
final SocketChannel socketChannel = (SocketChannel) key.channel();
|
||||
try{
|
||||
handler.handle(socketChannel);
|
||||
} catch (Exception e){
|
||||
throw new SocketRuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 实现写逻辑<br>
|
||||
* 当收到写出准备就绪的信号后，回调此方法，用户可向客户端发送消息
|
||||
*
|
||||
* @param datas 发送的数据
|
||||
* @return this
|
||||
*/
|
||||
public NioClient write(ByteBuffer... datas) {
|
||||
try {
|
||||
this.channel.write(datas);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取SocketChannel
|
||||
*
|
||||
* @return SocketChannel
|
||||
* @since 5.3.10
|
||||
*/
|
||||
public SocketChannel getChannel() {
|
||||
return this.channel;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
IoUtil.close(this.selector);
|
||||
IoUtil.close(this.channel);
|
||||
}
|
||||
}
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.core.thread.ThreadUtil;
|
||||
import cn.hutool.socket.SocketRuntimeException;
|
||||
|
||||
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.SocketChannel;
|
||||
import java.util.Iterator;
|
||||
|
||||
/**
|
||||
* NIO客户端
|
||||
*
|
||||
* @author looly
|
||||
* @since 4.4.5
|
||||
*/
|
||||
public class NioClient implements Closeable {
|
||||
|
||||
private Selector selector;
|
||||
private SocketChannel channel;
|
||||
private ChannelHandler handler;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param host 服务器地址
|
||||
* @param port 端口
|
||||
*/
|
||||
public NioClient(String host, int port) {
|
||||
init(new InetSocketAddress(host, port));
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param address 服务器地址
|
||||
*/
|
||||
public NioClient(InetSocketAddress address) {
|
||||
init(address);
|
||||
}
|
||||
|
||||
/**
|
||||
* 初始化
|
||||
*
|
||||
* @param address 地址和端口
|
||||
* @return this
|
||||
*/
|
||||
public NioClient init(InetSocketAddress address) {
|
||||
try {
|
||||
//创建一个SocketChannel对象，配置成非阻塞模式
|
||||
this.channel = SocketChannel.open();
|
||||
channel.configureBlocking(false);
|
||||
channel.connect(address);
|
||||
|
||||
//创建一个选择器，并把SocketChannel交给selector对象
|
||||
this.selector = Selector.open();
|
||||
channel.register(this.selector, SelectionKey.OP_READ);
|
||||
|
||||
// 等待建立连接
|
||||
//noinspection StatementWithEmptyBody
|
||||
while (false == channel.finishConnect()){}
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置NIO数据处理器
|
||||
*
|
||||
* @param handler {@link ChannelHandler}
|
||||
* @return this
|
||||
*/
|
||||
public NioClient setChannelHandler(ChannelHandler handler){
|
||||
this.handler = handler;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*/
|
||||
public void listen() {
|
||||
ThreadUtil.execute(() -> {
|
||||
try {
|
||||
doListen();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
private void doListen() throws IOException {
|
||||
while (this.selector.isOpen() && 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) {
|
||||
// 读事件就绪
|
||||
if (key.isReadable()) {
|
||||
final SocketChannel socketChannel = (SocketChannel) key.channel();
|
||||
try{
|
||||
handler.handle(socketChannel);
|
||||
} catch (Exception e){
|
||||
throw new SocketRuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 实现写逻辑<br>
|
||||
* 当收到写出准备就绪的信号后，回调此方法，用户可向客户端发送消息
|
||||
*
|
||||
* @param datas 发送的数据
|
||||
* @return this
|
||||
*/
|
||||
public NioClient write(ByteBuffer... datas) {
|
||||
try {
|
||||
this.channel.write(datas);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取SocketChannel
|
||||
*
|
||||
* @return SocketChannel
|
||||
* @since 5.3.10
|
||||
*/
|
||||
public SocketChannel getChannel() {
|
||||
return this.channel;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
IoUtil.close(this.selector);
|
||||
IoUtil.close(this.channel);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,153 +1,153 @@
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.log.Log;
|
||||
import cn.hutool.log.StaticLog;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.channels.SelectionKey;
|
||||
import java.nio.channels.Selector;
|
||||
import java.nio.channels.ServerSocketChannel;
|
||||
import java.nio.channels.SocketChannel;
|
||||
import java.util.Iterator;
|
||||
|
||||
/**
|
||||
* 基于NIO的Socket服务端实现
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class NioServer implements Closeable {
|
||||
private static final Log log = Log.get();
|
||||
|
||||
private static final AcceptHandler ACCEPT_HANDLER = new AcceptHandler();
|
||||
|
||||
private Selector selector;
|
||||
private ServerSocketChannel serverSocketChannel;
|
||||
private ChannelHandler handler;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param port 端口
|
||||
*/
|
||||
public NioServer(int port) {
|
||||
init(new InetSocketAddress(port));
|
||||
}
|
||||
|
||||
/**
|
||||
* 初始化
|
||||
*
|
||||
* @param address 地址和端口
|
||||
* @return this
|
||||
*/
|
||||
public NioServer init(InetSocketAddress address) {
|
||||
try {
|
||||
// 打开服务器套接字通道
|
||||
this.serverSocketChannel = ServerSocketChannel.open();
|
||||
// 设置为非阻塞状态
|
||||
this.serverSocketChannel.configureBlocking(false);
|
||||
// 绑定端口号
|
||||
this.serverSocketChannel.bind(address);
|
||||
|
||||
// 打开一个选择器
|
||||
this.selector = Selector.open();
|
||||
// 服务器套接字注册到Selector中 并指定Selector监控连接事件
|
||||
this.serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
|
||||
log.debug("Server listen on: [{}]...", address);
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置NIO数据处理器
|
||||
*
|
||||
* @param handler {@link ChannelHandler}
|
||||
* @return this
|
||||
*/
|
||||
public NioServer setChannelHandler(ChannelHandler handler){
|
||||
this.handler = handler;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取{@link Selector}
|
||||
*
|
||||
* @return {@link Selector}
|
||||
*/
|
||||
public Selector getSelector(){
|
||||
return this.selector;
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动NIO服务端，即开始监听
|
||||
*
|
||||
* @see #listen()
|
||||
*/
|
||||
public void start(){
|
||||
listen();
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*/
|
||||
public void listen() {
|
||||
try {
|
||||
doListen();
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
private void doListen() throws IOException {
|
||||
while (this.selector.isOpen() && 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) {
|
||||
// 有客户端接入此服务端
|
||||
if (key.isAcceptable()) {
|
||||
ACCEPT_HANDLER.completed((ServerSocketChannel) key.channel(), this);
|
||||
}
|
||||
|
||||
// 读事件就绪
|
||||
if (key.isReadable()) {
|
||||
final SocketChannel socketChannel = (SocketChannel) key.channel();
|
||||
try{
|
||||
handler.handle(socketChannel);
|
||||
} catch (Exception e){
|
||||
IoUtil.close(socketChannel);
|
||||
StaticLog.error(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
IoUtil.close(this.selector);
|
||||
IoUtil.close(this.serverSocketChannel);
|
||||
}
|
||||
}
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.log.Log;
|
||||
import cn.hutool.log.StaticLog;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.channels.SelectionKey;
|
||||
import java.nio.channels.Selector;
|
||||
import java.nio.channels.ServerSocketChannel;
|
||||
import java.nio.channels.SocketChannel;
|
||||
import java.util.Iterator;
|
||||
|
||||
/**
|
||||
* 基于NIO的Socket服务端实现
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class NioServer implements Closeable {
|
||||
private static final Log log = Log.get();
|
||||
|
||||
private static final AcceptHandler ACCEPT_HANDLER = new AcceptHandler();
|
||||
|
||||
private Selector selector;
|
||||
private ServerSocketChannel serverSocketChannel;
|
||||
private ChannelHandler handler;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param port 端口
|
||||
*/
|
||||
public NioServer(int port) {
|
||||
init(new InetSocketAddress(port));
|
||||
}
|
||||
|
||||
/**
|
||||
* 初始化
|
||||
*
|
||||
* @param address 地址和端口
|
||||
* @return this
|
||||
*/
|
||||
public NioServer init(InetSocketAddress address) {
|
||||
try {
|
||||
// 打开服务器套接字通道
|
||||
this.serverSocketChannel = ServerSocketChannel.open();
|
||||
// 设置为非阻塞状态
|
||||
this.serverSocketChannel.configureBlocking(false);
|
||||
// 绑定端口号
|
||||
this.serverSocketChannel.bind(address);
|
||||
|
||||
// 打开一个选择器
|
||||
this.selector = Selector.open();
|
||||
// 服务器套接字注册到Selector中 并指定Selector监控连接事件
|
||||
this.serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
|
||||
log.debug("Server listen on: [{}]...", address);
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置NIO数据处理器
|
||||
*
|
||||
* @param handler {@link ChannelHandler}
|
||||
* @return this
|
||||
*/
|
||||
public NioServer setChannelHandler(ChannelHandler handler){
|
||||
this.handler = handler;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取{@link Selector}
|
||||
*
|
||||
* @return {@link Selector}
|
||||
*/
|
||||
public Selector getSelector(){
|
||||
return this.selector;
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动NIO服务端，即开始监听
|
||||
*
|
||||
* @see #listen()
|
||||
*/
|
||||
public void start(){
|
||||
listen();
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*/
|
||||
public void listen() {
|
||||
try {
|
||||
doListen();
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
private void doListen() throws IOException {
|
||||
while (this.selector.isOpen() && 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) {
|
||||
// 有客户端接入此服务端
|
||||
if (key.isAcceptable()) {
|
||||
ACCEPT_HANDLER.completed((ServerSocketChannel) key.channel(), this);
|
||||
}
|
||||
|
||||
// 读事件就绪
|
||||
if (key.isReadable()) {
|
||||
final SocketChannel socketChannel = (SocketChannel) key.channel();
|
||||
try{
|
||||
handler.handle(socketChannel);
|
||||
} catch (Exception e){
|
||||
IoUtil.close(socketChannel);
|
||||
StaticLog.error(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
IoUtil.close(this.selector);
|
||||
IoUtil.close(this.serverSocketChannel);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,37 +1,37 @@
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.SelectableChannel;
|
||||
import java.nio.channels.Selector;
|
||||
|
||||
/**
|
||||
* NIO工具类
|
||||
*
|
||||
* @since 5.4.0
|
||||
*/
|
||||
public class NioUtil {
|
||||
|
||||
/**
|
||||
* 注册通道的指定操作到指定Selector上
|
||||
*
|
||||
* @param selector Selector
|
||||
* @param channel 通道
|
||||
* @param ops 注册的通道监听（操作）类型
|
||||
*/
|
||||
public static void registerChannel(Selector selector, SelectableChannel channel, Operation ops) {
|
||||
if (channel == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
channel.configureBlocking(false);
|
||||
// 注册通道
|
||||
//noinspection MagicConstant
|
||||
channel.register(selector, ops.getValue());
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.SelectableChannel;
|
||||
import java.nio.channels.Selector;
|
||||
|
||||
/**
|
||||
* NIO工具类
|
||||
*
|
||||
* @since 5.4.0
|
||||
*/
|
||||
public class NioUtil {
|
||||
|
||||
/**
|
||||
* 注册通道的指定操作到指定Selector上
|
||||
*
|
||||
* @param selector Selector
|
||||
* @param channel 通道
|
||||
* @param ops 注册的通道监听（操作）类型
|
||||
*/
|
||||
public static void registerChannel(Selector selector, SelectableChannel channel, Operation ops) {
|
||||
if (channel == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
channel.configureBlocking(false);
|
||||
// 注册通道
|
||||
//noinspection MagicConstant
|
||||
channel.register(selector, ops.getValue());
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,48 +1,48 @@
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import java.nio.channels.SelectionKey;
|
||||
|
||||
/**
|
||||
* SelectionKey Operation的枚举封装
|
||||
*
|
||||
* @author looly
|
||||
*/
|
||||
public enum Operation {
|
||||
|
||||
/** 读操作 */
|
||||
READ(SelectionKey.OP_READ),
|
||||
/** 写操作 */
|
||||
WRITE(SelectionKey.OP_WRITE),
|
||||
/** 连接操作 */
|
||||
CONNECT(SelectionKey.OP_CONNECT),
|
||||
/** 接受连接操作 */
|
||||
ACCEPT(SelectionKey.OP_ACCEPT);
|
||||
|
||||
private final int value;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param value 值
|
||||
* @see SelectionKey#OP_READ
|
||||
* @see SelectionKey#OP_WRITE
|
||||
* @see SelectionKey#OP_CONNECT
|
||||
* @see SelectionKey#OP_ACCEPT
|
||||
*/
|
||||
Operation(int value) {
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取值
|
||||
*
|
||||
* @return 值
|
||||
* @see SelectionKey#OP_READ
|
||||
* @see SelectionKey#OP_WRITE
|
||||
* @see SelectionKey#OP_CONNECT
|
||||
* @see SelectionKey#OP_ACCEPT
|
||||
*/
|
||||
public int getValue() {
|
||||
return this.value;
|
||||
}
|
||||
}
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import java.nio.channels.SelectionKey;
|
||||
|
||||
/**
|
||||
* SelectionKey Operation的枚举封装
|
||||
*
|
||||
* @author looly
|
||||
*/
|
||||
public enum Operation {
|
||||
|
||||
/** 读操作 */
|
||||
READ(SelectionKey.OP_READ),
|
||||
/** 写操作 */
|
||||
WRITE(SelectionKey.OP_WRITE),
|
||||
/** 连接操作 */
|
||||
CONNECT(SelectionKey.OP_CONNECT),
|
||||
/** 接受连接操作 */
|
||||
ACCEPT(SelectionKey.OP_ACCEPT);
|
||||
|
||||
private final int value;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param value 值
|
||||
* @see SelectionKey#OP_READ
|
||||
* @see SelectionKey#OP_WRITE
|
||||
* @see SelectionKey#OP_CONNECT
|
||||
* @see SelectionKey#OP_ACCEPT
|
||||
*/
|
||||
Operation(int value) {
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取值
|
||||
*
|
||||
* @return 值
|
||||
* @see SelectionKey#OP_READ
|
||||
* @see SelectionKey#OP_WRITE
|
||||
* @see SelectionKey#OP_CONNECT
|
||||
* @see SelectionKey#OP_ACCEPT
|
||||
*/
|
||||
public int getValue() {
|
||||
return this.value;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
/**
|
||||
* NIO相关封装
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
/**
|
||||
* NIO相关封装
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
package cn.hutool.socket.nio;
|
||||
@@ -1,7 +1,7 @@
|
||||
/**
|
||||
* Socket套接字相关工具类封装
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
/**
|
||||
* Socket套接字相关工具类封装
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
package cn.hutool.socket;
|
||||
@@ -1,24 +1,24 @@
|
||||
package cn.hutool.socket.protocol;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import cn.hutool.socket.aio.AioSession;
|
||||
|
||||
/**
|
||||
* 消息解码器
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
* @param <T> 解码后的目标类型
|
||||
*/
|
||||
public interface MsgDecoder<T> {
|
||||
/**
|
||||
* 对于从Socket流中获取到的数据采用当前MsgDecoder的实现类协议进行解析。
|
||||
*
|
||||
*
|
||||
* @param session 本次需要解码的session
|
||||
* @param readBuffer 待处理的读buffer
|
||||
* @return 本次解码成功后封装的业务消息对象, 返回null则表示解码未完成
|
||||
*/
|
||||
T decode(AioSession session, ByteBuffer readBuffer);
|
||||
}
|
||||
package cn.hutool.socket.protocol;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import cn.hutool.socket.aio.AioSession;
|
||||
|
||||
/**
|
||||
* 消息解码器
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
* @param <T> 解码后的目标类型
|
||||
*/
|
||||
public interface MsgDecoder<T> {
|
||||
/**
|
||||
* 对于从Socket流中获取到的数据采用当前MsgDecoder的实现类协议进行解析。
|
||||
*
|
||||
*
|
||||
* @param session 本次需要解码的session
|
||||
* @param readBuffer 待处理的读buffer
|
||||
* @return 本次解码成功后封装的业务消息对象, 返回null则表示解码未完成
|
||||
*/
|
||||
T decode(AioSession session, ByteBuffer readBuffer);
|
||||
}
|
||||
|
||||
@@ -1,23 +1,23 @@
|
||||
package cn.hutool.socket.protocol;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import cn.hutool.socket.aio.AioSession;
|
||||
|
||||
/**
|
||||
* 消息编码器
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
* @param <T> 编码前后的数据类型
|
||||
*/
|
||||
public interface MsgEncoder<T> {
|
||||
/**
|
||||
* 编码数据用于写出
|
||||
*
|
||||
* @param session 本次需要解码的session
|
||||
* @param writeBuffer 待处理的读buffer
|
||||
* @param data 写出的数据
|
||||
*/
|
||||
void encode(AioSession session, ByteBuffer writeBuffer, T data);
|
||||
}
|
||||
package cn.hutool.socket.protocol;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import cn.hutool.socket.aio.AioSession;
|
||||
|
||||
/**
|
||||
* 消息编码器
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
* @param <T> 编码前后的数据类型
|
||||
*/
|
||||
public interface MsgEncoder<T> {
|
||||
/**
|
||||
* 编码数据用于写出
|
||||
*
|
||||
* @param session 本次需要解码的session
|
||||
* @param writeBuffer 待处理的读buffer
|
||||
* @param data 写出的数据
|
||||
*/
|
||||
void encode(AioSession session, ByteBuffer writeBuffer, T data);
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
/**
|
||||
* 消息协议接口及实现
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
/**
|
||||
* 消息协议接口及实现
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
package cn.hutool.socket.protocol;
|
||||
Reference in New Issue
Block a user