Java使用Netty实现同时多端口监听
作者：齐 飞
本文主要内容为SpringBoot项目使用Netty监听多个端口接受客户端数据_netty监听多个端口,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
前言
Netty 是一个基于 Java NIO（非阻塞 I/O）的网络应用框架，它简化了开发高性能、高可靠性网络服务器和客户端的过程，使得开发者能够专注于业务逻辑的实现，而无需过多关注底层网络通信的复杂细节。适用于 Web 服务器、即时通讯、游戏服务器等开发，广泛应用于诸多领域。
环境
JDK：64位 Jdk1.8
SpringBoot：2.1.7.RELEASE
Netty：4.1.39.Final
实现功能
使用Netty监听多个端口接受设备发送的数据，并发送数据给客户端。
服务端
相关配置
pom.xml
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.39.Final</version>
</dependency>
application.yml
spring:
profiles:
# 环境 dev|test|prod
active: dev
netty-socket:
ports: 8801,8802
bufferSize: 2048
配置类
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
/**
* @author qf
* @since 2024/12/08 21:08
*/
@Component
@ConfigurationProperties(prefix = "netty-socket")
@Data
public class SocketConfig {
private List<Integer> ports;
private Integer bufferSize;
}
核心代码
启动类
CommandLineRunner：当应用程序启动时，CommandLineRunner 接口的实现类中的 run 方法会被调用
import cn.hutool.core.util.ObjectUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
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.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
@Slf4j
@Component
public class NettyServer {
@Value("${spring.profiles.active}")
private String active;
@Autowired
private SocketConfig socketConfig;
//用于关闭
private List<ChannelFuture> futures;
private EventLoopGroup bossGroup = null;
private EventLoopGroup workerGroup = null;
public List<ChannelFuture> getFutures() {
if (ObjectUtil.isEmpty(futures)) {
futures = new ArrayList<>();
}
return futures;
}
public void start() throws Exception {
bossGroup = new NioEventLoopGroup(1);
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))
.childHandler(new ListenChannelInitHandler());
if("dev".equals(active)){
bootstrap.handler(new LoggingHandler(LogLevel.INFO));
}
for (Integer port : socketConfig.getPorts()) {
ChannelFuture channelFuture = bootstrap.bind(port).sync();
getFutures().add(channelFuture);
channelFuture.addListener(future -> {
if (future.isSuccess()) {
log.info("端口：{} 监听成功！", port);
} else {
log.info("端口：{} 监听失败！", port);
}
});
channelFuture.channel().closeFuture()
.addListener((ChannelFutureListener) listener -> channelFuture.channel().close());
}
} catch (Exception e) {
log.info("netty监听数据时发生异常：", e);
} finally {
//异步使用stop()关闭
// bossGroup.shutdownGracefully();
// workerGroup.shutdownGracefully();
}
}
@PreDestroy
public void stop() {
if (ObjectUtil.isNotEmpty(futures)) {
try {
for (ChannelFuture future : futures) {
future.channel().close().addListener(ChannelFutureListener.CLOSE);
future.awaitUninterruptibly();
}
if (ObjectUtil.isNotEmpty(bossGroup) && ObjectUtil.isNotEmpty(workerGroup)) {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
futures = null;
log.info(" netty服务关闭成功！ ");
} catch (Exception e) {
log.error("关闭netty服务时发生异常: ", e);
}
}
}
}
端口枚举
public enum PortEnum {
TEST1(8801,"测试1"),
TEST2(8802,"测试2");
private int port;
private String name;
PortEnum(int port, String name) {
this.port = port;
this.name = name;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
端口处理器
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;
import static com.qf.netty.PortEnum.*;
/**
* 多端口处理器
*
* @author qf
* @since 2024/12/05 19:05
*/
@Slf4j
public class ListenChannelInitHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
int port = socketChannel.localAddress().getPort();
if (TEST1.getPort() == port) {
//使用自定义分割符
customizeSplitHandler(socketChannel, true, "$");
socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(new StringEncoder());
// 添加自定义的业务处理器
socketChannel.pipeline().addLast(new DeviceAServiceHandler());
} else if (TEST2.getPort() == port) {
customizeSplitHandler(socketChannel, false, "#");
socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(new StringEncoder());
socketChannel.pipeline().addLast(new DeviceBServiceHandler());
} else {
log.error("监听的端口号暂无业务处理：{}", port);
}
// 添加异常处理器
socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// 获取远程地址
InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIP = remoteAddress.getAddress().getHostAddress();
// 记录异常信息和客户端IP地址
log.error("监听设备时出现异常，客户端IP: {}", clientIP, cause);
ctx.close();
}
});
}
/**
* 自定义分隔符处理器
*
* @param socketChannel socketChannel
* @param stripDelimiter 是否去除分隔符
* @param split 分隔符
*/
private static void customizeSplitHandler(SocketChannel socketChannel, boolean stripDelimiter, String split) {
ByteBuf buffer = Unpooled.copiedBuffer(split.getBytes());
try {
socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(2048, stripDelimiter, buffer));
} catch (Exception e) {
buffer.release();
log.error("监听工地微站设备时出现异常：", e);
}
}
}
业务处理器
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
/**
* @author qf
* @since 2024/12/09 19:36
*/
@Slf4j
public class DeviceAServiceHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
// 处理接收到的消息
// 你可以在这里添加更复杂的业务逻辑，比如解析消息、访问数据库等。
log.info("发送的数据为------->：" + msg);
if (true) {
//发送数据给客户端
ctx.writeAndFlush("helloA~");
}
}
}
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
/**
* @author qf
* @since 2024/12/09 19:36
*/
@Slf4j
public class DeviceBServiceHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
// 处理接收到的消息
// 你可以在这里添加更复杂的业务逻辑，比如解析消息、访问数据库等。
log.info("发送的数据为------->：" + msg);
if (true) {
//发送数据给客户端
ctx.writeAndFlush("helloB~");
}
}
}
客户端
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 Hello1Client {
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", 8801))
.sync()//sync()是一个阻塞方法，只有连接建立后才会继续执行
.channel()//.channel()表示拿到服务器和客户端之间的SocketChannel(连接对象)
// 6. 向服务器发送数据
.writeAndFlush("hello, world$");
}
}
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;
public class Hello2Client {
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", 8802))
.sync()//sync()是一个阻塞方法，只有连接建立后才会继续执行
.channel()//.channel()表示拿到服务器和客户端之间的SocketChannel(连接对象)
// 6. 向服务器发送数据
.writeAndFlush("hello, world#");
}
}
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));
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
ctx.writeAndFlush(message);
}
@Override
public 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();
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// Close the connection when an exception is raised.
cause.printStackTrace();
ctx.close();
}
}
到此这篇关于Java使用Netty实现同时多端口监听的文章就介绍到这了,更多相关Java Netty 多端口监听内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！