mirror of
https://gitee.com/chinabugotech/hutool.git
synced 2025-07-21 15:09:48 +08:00
clean history
This commit is contained in:
Looly
117
hutool-socket/src/main/java/cn/hutool/socket/SocketConfig.java
Normal file
117
hutool-socket/src/main/java/cn/hutool/socket/SocketConfig.java
Normal file
@@ -0,0 +1,117 @@
|
||||
package cn.hutool.socket;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
|
||||
/**
|
||||
* Socket通讯配置
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class SocketConfig implements Serializable{
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
/** CPU核心数 */
|
||||
private static int CPU_COUNT = Runtime.getRuntime().availableProcessors();
|
||||
|
||||
/** 共享线程池大小，此线程池用于接收和处理用户连接 */
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +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);
|
||||
}
|
||||
}
|
||||
46
hutool-socket/src/main/java/cn/hutool/socket/SocketUtil.java
Normal file
46
hutool-socket/src/main/java/cn/hutool/socket/SocketUtil.java
Normal file
@@ -0,0 +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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +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);
|
||||
}
|
||||
|
||||
}
|
||||
138
hutool-socket/src/main/java/cn/hutool/socket/aio/AioClient.java
Normal file
138
hutool-socket/src/main/java/cn/hutool/socket/aio/AioClient.java
Normal file
@@ -0,0 +1,138 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
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 {
|
||||
|
||||
private 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参数
|
||||
* @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;
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到服务端
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioClient write(ByteBuffer data) {
|
||||
this.session.write(data);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭客户端
|
||||
*/
|
||||
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
|
||||
}
|
||||
186
hutool-socket/src/main/java/cn/hutool/socket/aio/AioServer.java
Normal file
186
hutool-socket/src/main/java/cn/hutool/socket/aio/AioServer.java
Normal file
@@ -0,0 +1,186 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
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;
|
||||
|
||||
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;
|
||||
|
||||
/**
|
||||
* 基于AIO的Socket服务端实现
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class AioServer {
|
||||
private static final Log log = LogFactory.get();
|
||||
private static AcceptHandler ACCEPT_HANDLER = new AcceptHandler();
|
||||
|
||||
private AsynchronousChannelGroup group;
|
||||
private AsynchronousServerSocketChannel channel;
|
||||
protected IoAction<ByteBuffer> ioAction;
|
||||
protected 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) {
|
||||
try {
|
||||
doStart(sync);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置 Socket 的 Option 选项<br>
|
||||
* 选项见：{@link java.net.StandardSocketOptions}
|
||||
*
|
||||
* @param <T> 选项泛型
|
||||
* @param name {@link SocketOption} 枚举
|
||||
* @param value SocketOption参数
|
||||
* @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) ? false : this.channel.isOpen();
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭服务
|
||||
*/
|
||||
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 是否阻塞
|
||||
* @throws IOException IO异常
|
||||
*/
|
||||
private void doStart(boolean sync) throws IOException {
|
||||
log.debug("Aio Server started, waiting for accept.");
|
||||
|
||||
// 接收客户端连接
|
||||
accept();
|
||||
|
||||
if (sync) {
|
||||
ThreadUtil.sync(this);
|
||||
}
|
||||
}
|
||||
// ------------------------------------------------------------------------------------- Private method end
|
||||
}
|
||||
206
hutool-socket/src/main/java/cn/hutool/socket/aio/AioSession.java
Normal file
206
hutool-socket/src/main/java/cn/hutool/socket/aio/AioSession.java
Normal file
@@ -0,0 +1,206 @@
|
||||
package cn.hutool.socket.aio;
|
||||
|
||||
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;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
import cn.hutool.socket.SocketConfig;
|
||||
import cn.hutool.socket.SocketUtil;
|
||||
|
||||
/**
|
||||
* AIO会话<br>
|
||||
* 每个客户端对应一个会话对象
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public class AioSession {
|
||||
|
||||
private static final ReadHandler READ_HANDLER = new ReadHandler();
|
||||
|
||||
private AsynchronousSocketChannel channel;
|
||||
private IoAction<ByteBuffer> ioAction;
|
||||
private ByteBuffer readBuffer;
|
||||
private ByteBuffer writeBuffer;
|
||||
/** 读取超时时长，小于等于0表示默认 */
|
||||
private long readTimeout;
|
||||
/** 写出超时时长，小于等于0表示默认 */
|
||||
private long writeTimeout;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param channel {@link AsynchronousSocketChannel}
|
||||
* @param ioAction IO消息处理类
|
||||
* @param config 配置项
|
||||
*/
|
||||
public AioSession(AsynchronousSocketChannel channel, IoAction<ByteBuffer> ioAction, SocketConfig config) {
|
||||
this.channel = channel;
|
||||
this.readBuffer = ByteBuffer.allocate(config.getReadBufferSize());
|
||||
this.writeBuffer = ByteBuffer.allocate(config.getWriteBufferSize());
|
||||
this.ioAction = ioAction;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取{@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;
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到目标端，并关闭输出
|
||||
*
|
||||
* @return this
|
||||
*/
|
||||
public AioSession writeAndClose(ByteBuffer data) {
|
||||
write(data);
|
||||
return closeOut();
|
||||
}
|
||||
|
||||
/**
|
||||
* 写数据到目标端
|
||||
*
|
||||
* @return {@link Future}
|
||||
*/
|
||||
public Future<Integer> write(ByteBuffer data) {
|
||||
return this.channel.write(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) ? false : 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;
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭会话
|
||||
*/
|
||||
public void close() {
|
||||
IoUtil.close(this.channel);
|
||||
this.readBuffer = null;
|
||||
this.writeBuffer = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行读，用于读取事件结束的回调
|
||||
*/
|
||||
protected void callbackRead() {
|
||||
readBuffer.flip();// 读模式
|
||||
ioAction.doAction(this, readBuffer);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +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);
|
||||
}
|
||||
@@ -0,0 +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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
/**
|
||||
* AIO相关封装
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
package cn.hutool.socket.aio;
|
||||
@@ -0,0 +1,83 @@
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.SocketChannel;
|
||||
|
||||
import cn.hutool.core.io.IORuntimeException;
|
||||
|
||||
/**
|
||||
* NIO客户端
|
||||
*
|
||||
* @author looly
|
||||
* @since 4.4.5
|
||||
*/
|
||||
public class NioClient {
|
||||
|
||||
private SocketChannel channel;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @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 {
|
||||
this.channel = SocketChannel.open(address);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理读事件<br>
|
||||
* 当收到读取准备就绪的信号后，回调此方法，用户可读取从客户端传世来的消息
|
||||
*
|
||||
* @param buffer 服务端数据存储缓存
|
||||
*/
|
||||
public NioClient read(ByteBuffer buffer) {
|
||||
try {
|
||||
this.channel.read(buffer);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 实现写逻辑<br>
|
||||
* 当收到写出准备就绪的信号后，回调此方法，用户可向客户端发送消息
|
||||
*
|
||||
* @param datas 发送的数据
|
||||
*/
|
||||
public NioClient write(ByteBuffer... datas) {
|
||||
try {
|
||||
this.channel.write(datas);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
}
|
||||
174
hutool-socket/src/main/java/cn/hutool/socket/nio/NioServer.java
Normal file
174
hutool-socket/src/main/java/cn/hutool/socket/nio/NioServer.java
Normal file
@@ -0,0 +1,174 @@
|
||||
package cn.hutool.socket.nio;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.ServerSocket;
|
||||
import java.nio.channels.SelectableChannel;
|
||||
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 cn.hutool.core.io.IORuntimeException;
|
||||
import cn.hutool.core.io.IoUtil;
|
||||
|
||||
/**
|
||||
* 基于NIO的Socket服务端实现
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
public abstract class NioServer implements Closeable {
|
||||
|
||||
private Selector selector;
|
||||
private ServerSocketChannel serverSocketChannel;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param port 端口
|
||||
*/
|
||||
public NioServer(int port) {
|
||||
init(new InetSocketAddress(port));
|
||||
}
|
||||
|
||||
/**
|
||||
* 初始化
|
||||
*
|
||||
* @param address 地址和端口
|
||||
* @return this
|
||||
*/
|
||||
public NioServer init(InetSocketAddress address) {
|
||||
try {
|
||||
// 打开服务器套接字通道
|
||||
this.serverSocketChannel = ServerSocketChannel.open();
|
||||
// 设置为非阻塞状态
|
||||
serverSocketChannel.configureBlocking(false);
|
||||
// 获取通道相关联的套接字
|
||||
final ServerSocket serverSocket = serverSocketChannel.socket();
|
||||
// 绑定端口号
|
||||
serverSocket.bind(address);
|
||||
|
||||
// 打开一个选择器
|
||||
selector = Selector.open();
|
||||
// 服务器套接字注册到Selector中 并指定Selector监控连接事件
|
||||
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*/
|
||||
public void listen() {
|
||||
try {
|
||||
doListen();
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始监听
|
||||
*
|
||||
* @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) {
|
||||
// 有客户端接入此服务端
|
||||
if (key.isAcceptable()) {
|
||||
// 获取通道 转化为要处理的类型
|
||||
final ServerSocketChannel server = (ServerSocketChannel) key.channel();
|
||||
SocketChannel socketChannel;
|
||||
try {
|
||||
// 获取连接到此服务器的客户端通道
|
||||
socketChannel = server.accept();
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
|
||||
// SocketChannel通道的可读事件注册到Selector中
|
||||
registerChannel(selector, socketChannel, Operation.READ);
|
||||
}
|
||||
|
||||
// 读事件就绪
|
||||
if (key.isReadable()) {
|
||||
final SocketChannel socketChannel = (SocketChannel) key.channel();
|
||||
read(socketChannel);
|
||||
|
||||
// SocketChannel通道的可写事件注册到Selector中
|
||||
registerChannel(selector, socketChannel, Operation.WRITE);
|
||||
}
|
||||
|
||||
// 写事件就绪
|
||||
if (key.isWritable()) {
|
||||
final SocketChannel socketChannel = (SocketChannel) key.channel();
|
||||
write(socketChannel);
|
||||
// SocketChannel通道的可读事件注册到Selector中
|
||||
registerChannel(selector, socketChannel, Operation.READ);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
IoUtil.close(this.selector);
|
||||
IoUtil.close(this.serverSocketChannel);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理读事件<br>
|
||||
* 当收到读取准备就绪的信号后，回调此方法，用户可读取从客户端传世来的消息
|
||||
*
|
||||
* @param socketChannel SocketChannel
|
||||
*/
|
||||
protected abstract void read(SocketChannel socketChannel);
|
||||
|
||||
/**
|
||||
* 实现写逻辑<br>
|
||||
* 当收到写出准备就绪的信号后，回调此方法，用户可向客户端发送消息
|
||||
*
|
||||
* @param socketChannel SocketChannel
|
||||
*/
|
||||
protected abstract void write(SocketChannel socketChannel);
|
||||
|
||||
/**
|
||||
* 注册通道到指定Selector上
|
||||
*
|
||||
* @param selector Selector
|
||||
* @param channel 通道
|
||||
* @param ops 注册的通道监听类型
|
||||
*/
|
||||
private void registerChannel(Selector selector, SelectableChannel channel, Operation ops) {
|
||||
if (channel == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
channel.configureBlocking(false);
|
||||
// 注册通道
|
||||
channel.register(selector, ops.getValue());
|
||||
} catch (IOException e) {
|
||||
throw new IORuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +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 int value;
|
||||
|
||||
/**
|
||||
* 构造
|
||||
*
|
||||
* @param value 值
|
||||
* @see SelectionKey#OP_READ
|
||||
* @see SelectionKey#OP_WRITE
|
||||
* @see SelectionKey#OP_CONNECT
|
||||
* @see SelectionKey#OP_ACCEPT
|
||||
*/
|
||||
private 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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
/**
|
||||
* NIO相关封装
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
package cn.hutool.socket.nio;
|
||||
@@ -0,0 +1,7 @@
|
||||
/**
|
||||
* Socket套接字相关工具类封装
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
package cn.hutool.socket;
|
||||
@@ -0,0 +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);
|
||||
}
|
||||
@@ -0,0 +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);
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
/*
|
||||
* Copyright (c) 2017, org.smartboot. All rights reserved.
|
||||
* project name: smart-socket
|
||||
* file name: Protocol.java
|
||||
* Date: 2017-11-25
|
||||
* Author: sandao
|
||||
*/
|
||||
|
||||
package cn.hutool.socket.protocol;
|
||||
|
||||
/**
|
||||
* 协议接口<br>
|
||||
* 通过实现此接口完成消息的编码和解码
|
||||
*
|
||||
* <p>
|
||||
* 所有Socket使用相同协议对象，类成员变量和对象成员变量易造成并发读写问题。
|
||||
* </p>
|
||||
*
|
||||
* @author Looly
|
||||
*/
|
||||
public interface Protocol<T> extends MsgEncoder<T>, MsgDecoder<T> {
|
||||
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
/**
|
||||
* 消息协议接口及实现
|
||||
*
|
||||
* @author looly
|
||||
*
|
||||
*/
|
||||
package cn.hutool.socket.protocol;
|
||||
Reference in New Issue
Block a user