netty實現TCP長連接

所用jar包

netty-all-4.1.30.Final.jar 密碼：rzwe

NettyConfig.java，存放連接的客戶端

import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

public class NettyConfig {
    
    /**
     * 存儲每一個客戶端接入進來時的channel對象
     */
    public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
}

 

Server.java，netty配置信息

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class Server {
    private int port;
    private ServerSocketChannel serverSocketChannel;
 
    public Server(int port){
        this.port = port;
        bind();
    }
    
    private void bind() {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                //服務端要建立兩個group，一個負責接收客戶端的連接，一個負責處理數據傳輸
                //連接處理group
                EventLoopGroup boss = new NioEventLoopGroup();
                //事件處理group
                EventLoopGroup worker = new NioEventLoopGroup();
                ServerBootstrap bootstrap = new ServerBootstrap();
                // 綁定處理group
                bootstrap.group(boss, worker).channel(NioServerSocketChannel.class)
                        //保持連接數
                        .option(ChannelOption.SO_BACKLOG, 300)
                        //有數據立即發送
                        .option(ChannelOption.TCP_NODELAY, true)
                        //保持連接
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        //處理新連接
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel sc) throws Exception {
                                // 增加任務處理
                                ChannelPipeline p = sc.pipeline();
                                p.addLast(
//                                        //使用了netty自帶的編碼器和解碼器
//                                        new StringDecoder(),
//                                        new StringEncoder(),
                                        //心跳檢測，讀超時，寫超時，讀寫超時
                                        //new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS),
                                        //自定義的處理器
                                        new ServerHandler());
                            }
                        });
 
                //綁定端口，同步等待成功
                ChannelFuture future;
                try {
                    future = bootstrap.bind(port).sync();
                    if (future.isSuccess()) {
                        serverSocketChannel = (ServerSocketChannel) future.channel();
                        System.out.println("服務端啟動成功，端口："+port);
                    } else {
                        System.out.println("服務端啟動失敗！");
                    }
                    
                    //等待服務監聽端口關閉,就是由於這里會將線程阻塞，導致無法發送信息，所以我這里開了線程
                    future.channel().closeFuture().sync();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                finally {
                    //優雅地退出，釋放線程池資源
                    boss.shutdownGracefully();
                    worker.shutdownGracefully();
                }
            }
        });
        thread.start();
    }
    
    public void sendMessage(Object msg){
        if(serverSocketChannel != null){
            serverSocketChannel.writeAndFlush(msg);
        }
    }
    
    public static void main(String[] args) {
        Server server = new Server(8088);
    }
}

 

ServerHandler.java，業務處理

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class ServerHandler extends ChannelInboundHandlerAdapter {    
      
    /**
     * 客戶端與服務端創建連接的時候調用
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客戶端與服務端連接開始...");
        NettyConfig.group.add(ctx.channel());
    }
 
    /**
     * 客戶端與服務端斷開連接時調用
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客戶端與服務端連接關閉...");
        NettyConfig.group.remove(ctx.channel());
    }
 
    /**
     * 服務端接收客戶端發送過來的數據結束之后調用
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
        System.out.println("信息接收完畢...");
    }
 
    /**
     * 工程出現異常的時候調用
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
 
    /**
     * 服務端處理客戶端websocket請求的核心方法，這里接收了客戶端發來的信息
     */
    @Override
    public void channelRead(ChannelHandlerContext channelHandlerContext, Object info) throws Exception {
        System.out.println("接收到了："+info);
ByteBuf buf = (ByteBuf) info;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String(req, "UTF-8");
System.out.println("接收客戶端數據:" + body);
ByteBuf pingMessage = Unpooled.buffer();
pingMessage.writeBytes(req);
channelHandlerContext.writeAndFlush(pingMessage);

        
        //服務端使用這個就能向 每個連接上來的客戶端群發消息
        //NettyConfig.group.writeAndFlush(info);
//        Iterator<Channel> iterator = NettyConfig.group.iterator();
//        while(iterator.hasNext()){
//            //打印出所有客戶端的遠程地址
//            System.out.println((iterator.next()).remoteAddress());
//        }
    }
    
    
}

 

使用網絡調試助手進行連接測試 下載地址

https://www.wanpishe.top/detail?blogId=fc62fce2-020a-4815-8388-0903e4a54e1f