Spring实现Netty-TCP-client-server简单实例

2021-10-27

增加gradle依赖

1 2	// https://mvnrepository.com/artifact/io.netty/netty-all implementation group: 'io.netty', name: 'netty-all', version: '4.1.68.Final'

netty下server实现

NettyServer

包：netty.tcp.server

import com.cec.ice.netty.tcp.serverChannel.NettyServerChannel;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class NettyServer {

    // 服务器端口
    private int port = 8098;

    public void run() throws Exception {

        // NioEventLoopGroup是用来处理IO操作的多线程事件循环器
        // 用来接收进来的连接
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 用来处理已经被接收的连接
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            // 一个启动NIO服务的辅助启动类
            ServerBootstrap server = new ServerBootstrap();
            // 添加bossGroup，workerGroup
            server.group(bossGroup, workerGroup)
                    // Channel如何接收新的连接
                    .channel(NioServerSocketChannel.class)
                    // 自定义通道处理
                    .childHandler(new NettyServerChannel<SocketChannel>());

            // 有效socket的最大个数
            server.option(ChannelOption.SO_BACKLOG, 128);
            // TCP实现KEEPALIVE
            server.childOption(ChannelOption.SO_KEEPALIVE, true);
            // 绑定端口，开始接收进来的连接
            ChannelFuture f = server.bind(port).sync();

            System.out.println("server started successfully... \n");

            // 服务器关闭监听
            f.channel().closeFuture().sync();
        } finally {
            // 关闭EventLoopGroup，释放所有资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new NettyServer().run();
    }
}

NettyServerChannel

包：netty.tcp.serverChannel

import com.cec.ice.netty.tcp.serverHandler.NettyServerHandler;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;

public class NettyServerChannel<SocketChannel> extends ChannelInitializer<Channel> {

    @Override
    protected void initChannel(Channel ch) throws Exception {
        // 自定义处理类
        // 可自定义粘包处理
        ch.pipeline().addLast(new NettyServerHandler());
    }
}

NettyServerHandler

包：netty.tcp.serverHandler

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;

import java.net.InetSocketAddress;

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    // 建立连接时执行
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client's ID: " + ctx.channel().id());

        byte[] data = "hello, client! ".getBytes();
        sendMessage(ctx, data);
    }

    // 收到数据时调用
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            // 对象转字节缓冲区
            ByteBuf in = (ByteBuf) msg;
            // 字节缓冲区转字符串
            String revString = getMessage(in);

            // 差异化客户端
            InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
            // 当前连接的host
            String host = socketAddress.getAddress().getHostAddress();
            // 当前连接的端口
            int port = socketAddress.getPort();

            // 根据IP，PORT区别不同客户端，执行不同处理。
            System.out.println(host + ":" + port + "--->" + revString);

            // 发送数据到客户端
            byte[] data = "see you, client! ".getBytes();
            sendMessage(ctx, data);

        } finally {
            // 抛弃收到的数据
            ReferenceCountUtil.release(msg);
        }
    }

    // 捕获异常执行
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }

    // 通道连接断开执行
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 调用父类方法处理
        super.channelInactive(ctx);
        // 关闭通道
        ctx.close();
    }

    // 字节缓冲区转字符串
    public String getMessage(ByteBuf buf) {
        byte[] con = new byte[buf.readableBytes()];
        buf.readBytes(con);
        return new String(con);
    }

    // 发送字节数组到通道
    public void sendMessage(ChannelHandlerContext ctx, byte[] msg) {
        ByteBuf message = Unpooled.buffer(msg.length);
        message.writeBytes(msg);
        ctx.writeAndFlush(message);
    }
}

netty下client实现

NettyClient

包：netty.tcp.client

import com.cec.ice.netty.tcp.clientChannel.NettyClientChannel;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyClient {

    // 服务器IP
    private String host = "localhost";

    // 服务器端口号
    private int port = 8098;

    private void run() throws InterruptedException {

        // 初始化IO操作事件循环器
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

        try {
            // NIO服务的辅助启动类
            Bootstrap bootstrap = new Bootstrap();
            // Channel如何接收新的连接
            bootstrap.channel(NioSocketChannel.class);
            // TCP实现KEEPALIVE
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            // 添加eventLoopGroup
            bootstrap.group(eventLoopGroup);
            // 自定义通道处理
            bootstrap.handler(new NettyClientChannel<SocketChannel>());
            // 连接服务器
            ChannelFuture future = bootstrap.connect(host, port).sync();

            // 连接服务器成功
            if (future.isSuccess()) {
                System.out.println("connected server successfully! \n");
            } else {
                // 重连...
            }

            // 服务器关闭监听
            future.channel().closeFuture().sync();
        } finally {
            // 关闭EventLoopGroup，释放所有资源
            eventLoopGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new NettyClient().run();
    }
}

NettyClientChannel

包：netty.tcp.clientChannel

import com.cec.ice.netty.tcp.clientHandler.NettyClientHandler;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;

public class NettyClientChannel<SocketChannel> extends ChannelInitializer<Channel> {

    @Override
    protected void initChannel(Channel ch) throws Exception {
        // 自定义处理类
        // 可自定义粘包处理
        ch.pipeline().addLast(new NettyClientHandler());
    }
}

NettyClientHandler

包：netty.tcp.clientHandler

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    // 记录可用通道
    private ChannelHandlerContext ctx;

    // 连接成功后执行
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;

        byte[] data = "hello, server! ".getBytes();
        sendMessage(data);
    }

    // 通道有数据执行
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        String rev = getMessage(buf);

        System.out.println(rev);

        // 发送数据到服务端
        byte[] data = "see you, server! ".getBytes();
        sendMessage(data);
    }

    // 通道连接断开执行
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 调用父类方法处理
        super.channelInactive(ctx);
        // 关闭通道
        ctx.close();

        // 重连...
    }

    // 字节缓冲区转字符串
    public String getMessage(ByteBuf buf) {
        byte[] con = new byte[buf.readableBytes()];
        buf.readBytes(con);
        return new String(con);
    }

    // 发送字节数组到通道
    public void sendMessage(byte[] msg) {
        ByteBuf message = Unpooled.buffer(msg.length);
        message.writeBytes(msg);
        ctx.writeAndFlush(message);
    }
}

Run

先运行服务端

再运行客户端

服务端结果：

server started successfully...

client's ID: d2f7a0fb
127.0.0.1:xxxxx--->hello, server! 
127.0.0.1:xxxxx--->see you, server! 
......

客户端结果：

connected server successfully!

hello, client! 
see you, client! 
......