开发者

Java使用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))aEclSBDSW;
            }
            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().a编程客栈ddLast(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();
            lo编程客栈g.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()//syncjavascript()是一个阻塞方法，只有连接建立后才会继续执行
                .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(ChannelHandlerCophpntext 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 多端口监听内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！

    继续阅读：Java Netty 多端口监听netty监听多个端口

    更多精彩内容

    0

    上一篇:

    下一篇:

    精彩评论

    暂无评论...
    登录 注册
    Qedev.com
    验证码 换一张
    取 消

    最新开发

    开发排行榜

    法律声明：本站内容均为网友上传，网站举办方负责审核和监督，

    如存在版权或非法内容，欢迎举报，我们将尽快予以删除。

    Copyright © 2018-2020 开发者. All rights reserved.

    Powered by 开发者

    ICP备案号： 京ICP备10032868号-9