当前位置：首页 > news >正文

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

news 2025/7/27 12:46:19

文章目录

前言
服务端
- 相关配置
- 核心代码
客户端

前言

前言
环境：
JDK：64位 Jdk1.8
SpringBoot：2.1.7.RELEASE
Netty：4.1.39.Final

功能：
使用Netty监听端口接受客户端的数据，并发送数据给客户端。

服务端

核心代码

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:11*/
@Slf4j
@Component
public class CommandLineRunnerImpl implements CommandLineRunner {@Autowiredprivate NettyServer nettyServer;@Overridepublic void run(String... args) throws Exception {log.info("-----------监听端口启动成功！-----------");nettyServer.start(); // 启动Netty服务}
}

服务类

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;/*** @author qf* @since 2024/10/08 21:13*/
@Slf4j
@Component
public class NettyServer {@Autowiredprivate SocketConfig socketConfig;public void start() throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {Integer bufferSize = socketConfig.getBufferSize();ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(bufferSize, bufferSize, bufferSize)).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {//\n和\r\n分隔符ch.pipeline().addLast(new LineBasedFrameDecoder(bufferSize));ch.pipeline().addLast(new StringDecoder());ch.pipeline().addLast(new StringEncoder());// 添加自定义的业务处理器ch.pipeline().addLast(new DeviceServerHandler());}});ChannelFuture channelFuture = bootstrap.bind(socketConfig.getPort()).sync();log.info("启动TCP服务器启动成功,正在监听端口:{}", socketConfig.getPort());channelFuture.channel().closeFuture().sync();} catch (Exception e) {log.info("netty发生错误：", e);} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

业务处理器

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;/*** @author qf* @since 2024/10/08 21:15*/
@Slf4j
public class DeviceServerHandler extends SimpleChannelInboundHandler<String> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String dateStr) throws Exception {// 处理接收到的消息// 你可以在这里添加更复杂的业务逻辑，比如解析消息、访问数据库等。log.info("'监听的数据为'------->：" + dateStr);if (true) {//发送数据给客户端ctx.writeAndFlush("hello~");}}}

客户端

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;import java.net.InetSocketAddress;/*** @author qf* @since 2024/10/11 18:01*/
public class HelloClient {public static void main(String[] args) throws InterruptedException {// 1. 启动类，启动客户端new Bootstrap()// 2. 添加 EventLoop.group(new NioEventLoopGroup())//如果服务器端发来数据，客户端的EventLoop就可以从selector触发读事件进行处理// 3. 选择客户端 channel 实现，底层封装了SocketChannel.channel(NioSocketChannel.class)// 4. 添加处理器.handler(new ChannelInitializer<NioSocketChannel>() {@Override // 在连接建立后被调用protected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());//编码器，将字符串编码成ByteBuf进行发送ch.pipeline().addLast(new EchoClientHandler());}})// 5. 连接到服务器.connect(new InetSocketAddress("localhost", 9992)).sync()//sync()是一个阻塞方法，只有连接建立后才会继续执行.channel()//.channel()表示拿到服务器和客户端之间的SocketChannel(连接对象)// 6. 向服务器发送数据.writeAndFlush("hello, world\n");}
}

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;/*** @author qf* @since 2024/10/11 18:05*/
public class EchoClientHandler extends ChannelInboundHandlerAdapter {private final ByteBuf message;public EchoClientHandler() {message = Unpooled.buffer(256);message.writeBytes("hello netty".getBytes(CharsetUtil.UTF_8));}@Overridepublic void channelActive(ChannelHandlerContext ctx) {ctx.writeAndFlush(message);}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {String data = ((ByteBuf) msg).toString(CharsetUtil.UTF_8);System.out.println(data);
//        ctx.write(msg);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) {ctx.flush();}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {// Close the connection when an exception is raised.cause.printStackTrace();ctx.close();}
}