SpringBoot项目监听端口接受数据(NIO版)

---

前言

环境：
JDK：64位 Jdk1.8
SpringBoot：2.1.7.RELEASE

功能：
使用Java中原生的NIO监听端口接受客户端的数据，并发送数据给客户端。

服务端

相关配置

application.yml

socket:port: 9991bufferSize: 2048timeout: 3000

配置类

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;/*** @author qf* @since 2024/10/08 21:20*/
@Component
@ConfigurationProperties(prefix = "socket")
@Setter
@Getter
@ToString
public class NioSocketConfig {private Integer port;private Integer bufferSize;private Integer timeout;
}

核心代码

CommandLineRunner
当应用程序启动时，CommandLineRunner 接口的实现类中的 run 方法会被调用

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;/*** @author qf* @since 2024/10/08 21:25*/
@Slf4j
@Component
public class CommandLineRunnerImpl implements CommandLineRunner {private NioSocketConfig nioSocketConfig;@Autowiredpublic CommandLineRunnerImpl(NioSocketConfig nioSocketConfig) {this.nioSocketConfig = nioSocketConfig;}@Overridepublic void run(String... args) {ServerSelector serverSelector = new ServerSelector(nioSocketConfig);log.info("-----------监听端口启动成功！-----------");serverSelector.server();}
}

服务类

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;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.util.Iterator;/*** NIO Socket Server* @author qf* @since 2024/10/08 21:30*/
@Slf4j
public class ServerSelector {private NioSocketConfig nioSocketConfig;public ServerSelector(NioSocketConfig nioSocketConfig){this.nioSocketConfig = nioSocketConfig;}@Beanpublic void server() {Selector selector = null;Protocol protocol = null;try {// 实例化一个信道ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();// 将该信道绑定到指定端口serverSocketChannel.bind(new InetSocketAddress(nioSocketConfig.getPort()));// 配置信道为非阻塞模式serverSocketChannel.configureBlocking(false);// 创建一个选择器selector = Selector.open();// 将选择器注册到各个信道serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);// 初始化事件处理器protocol = new EchoSelectorProtocol(nioSocketConfig.getBufferSize());} catch (IOException e) {log.error("粒径设备监听10091端口时发生异常：",e);}// 不断轮询select方法，获取准备好的信道所关联的Key集while (true) {try {if (selector == null || protocol == null) {break;}Thread.sleep(100);// 一直等待,直至有信道准备好了I/O操作if (selector.select(nioSocketConfig.getTimeout()) == 0) {// 在等待信道准备的同时，也可以异步地执行其他任务，continue;}// 获取准备好的信道所关联的Key集合的iterator实例Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();// 循环取得集合中的每个键值while (iterator.hasNext()) {SelectionKey selectionKey = iterator.next();// 如果服务端信道感兴趣的I/O操作为acceptif (selectionKey.isValid() && selectionKey.isAcceptable()) {protocol.handleAccept(selectionKey);}// 如果客户端信道感兴趣的I/O操作为readif (selectionKey.isValid() && selectionKey.isReadable()) {protocol.handleRead(selectionKey);}// 如果该键值有效，并且其对应的客户端信道感兴趣的I/O操作为writeif (selectionKey.isValid() && selectionKey.isWritable()) {protocol.handleWrite(selectionKey);}// 这里需要手动从键集中移除当前的keyiterator.remove();}} catch (Exception e) {log.error("监听端口轮询selector时发生异常：",e);}}}
}

协议接口

import java.io.IOException;
import java.nio.channels.SelectionKey;/*** 该接口定义了通用TCPSelectorServer类与特定协议之间的接口，* 它把与具体协议相关的处理各种I/O的操作分离了出来，* 以使不同协议都能方便地使用这个基本的服务模式。* @author qf* @since 2024/10/08 20:30*/
public interface Protocol {//accept I/O形式void handleAccept(SelectionKey selectionKey) throws IOException;//read I/O形式void handleRead(SelectionKey selectionKey) throws IOException;//write I/O形式void handleWrite(SelectionKey selectionKey) throws IOException;
}

实现类

import lombok.extern.slf4j.Slf4j;import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;/*** @author qf* @since 2024/10/08 20:30*/
@Slf4j
public class EchoSelectorProtocol implements Protocol {private int bufSize; // 缓冲区的长度public EchoSelectorProtocol(int bufSize) {this.bufSize = bufSize;}// 服务端信道已经准备好了接收新的客户端连接public void handleAccept(SelectionKey selectionKey) {try {SocketChannel socketChannel = ((ServerSocketChannel) selectionKey.channel()).accept();socketChannel.configureBlocking(false);// 将选择器注册到连接到的客户端信道，并指定该信道key值的属性为OP_READ，同时为该信道指定关联的附件socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ, ByteBuffer.allocate(bufSize));} catch (IOException e) {log.error("accept异常：",e);selectionKey.cancel();}}// 客户端信道已经准备好了从信道中读取数据到缓冲区public void handleRead(SelectionKey selectionKey) {try {SocketChannel socketChannel = (SocketChannel) selectionKey.channel();if (!(selectionKey.attachment() instanceof ByteBuffer)) {return;}// 获取该信道所关联的附件，这里为缓冲区ByteBuffer byteBuffer = (ByteBuffer) selectionKey.attachment();long bytesRead = socketChannel.read(byteBuffer);// 如果read()方法返回-1，说明客户端关闭了连接，那么客户端已经接收到了与自己发送字节数相等的数据，可以安全地关闭if (bytesRead == -1) {socketChannel.close();} else if (bytesRead > 0) {// 将channel改为读取状态byteBuffer.flip();String dateStr = new String(byteBuffer.array());if (true) {// 注册写事件selectionKey.interestOps(SelectionKey.OP_WRITE);selectionKey.attach("test"); // 将数据附加到SelectionKey上}byteBuffer.clear();// 如果缓冲区总读入了数据，则将该信道感兴趣的操作设置为为可读可写selectionKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);}} catch (IOException e) {log.error("read异常，", e);selectionKey.cancel();} catch (Exception e) {log.error("read异常：", e);}}// 客户端信道已经准备好了将数据从缓冲区写入信道public void handleWrite(SelectionKey selectionKey) throws IOException {SocketChannel socketChannel = (SocketChannel) selectionKey.channel();if (selectionKey.attachment() instanceof String) {String data = (String) selectionKey.attachment(); // 获取附加的数据ByteBuffer buffer = ByteBuffer.wrap(data.getBytes(StandardCharsets.UTF_8));socketChannel.write(buffer);}// 写完后取消写事件，重新注册读事件selectionKey.interestOps(SelectionKey.OP_READ);}
}

客户端

import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;/** *  模拟客户端* @author qf* @date 2024/7/8 19:25 */
@Slf4j
public class NIOClient {public static void main(String[] args) throws Exception{//得到一个网络通道SocketChannel socketChannel = SocketChannel.open();//设置非阻塞socketChannel.configureBlocking(false);//提供服务器端的ip 和 端口InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 9991);//连接服务器if (!socketChannel.connect(inetSocketAddress)) {while (!socketChannel.finishConnect()) {//!没有 完成连接finishConnect方法log.info("因为连接需要时间，客户端不会阻塞，可以做其它工作..");}}//...如果连接成功，就发送数据String str  = "hello！";ByteBuffer buffer = ByteBuffer.wrap(str.getBytes());//发送数据，将 buffer 数据写入 channelsocketChannel.write(buffer);// 接收服务器发送的数据int count = 0;while (true) {ByteBuffer readBuffer = ByteBuffer.allocate(1024 * 1024);count += socketChannel.read(readBuffer);if(count != 0){System.out.println(count);String data = StandardCharsets.UTF_8.decode(readBuffer).toString();System.out.println(data);String x = new String(readBuffer.array());System.out.println(x);count = 0;}readBuffer.clear();}}
}

---

相关文章：
SpringBoot项目监听端口接受数据(Netty版)