Netty服务器实时通信实战

      io.netty
      netty-all
      4.1.49.Final

package websocker;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class WebSockerServer {
    public static void main(String[] args) {
        //创建一组线程池
        //主线程池：用于接受客户端的请求链接，不做任何处理
        NioEventLoopGroup group1 = new NioEventLoopGroup();
        //从线程池：主线程池会把任务交给它，让其做任务
        NioEventLoopGroup group2 = new NioEventLoopGroup();
        try{
            //创建服务器启动类
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(group1,group2)//设置主从线程池
                    .channel(NioServerSocketChannel.class)//设置nio双向通道
                    .childHandler(new WSServerInitializer());//添加子处理器，用于处理从线程池的任务
            //启动服务，并且设置端口号，启动方式为同步
            ChannelFuture channelFuture = serverBootstrap.bind(8800).sync();
            //监听关闭的channel，设置为同步方式
            channelFuture.channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            group1.shutdownGracefully();
            group2.shutdownGracefully();
        }
    }
}

package websocker;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WSServerInitializer extends ChannelInitializer {

    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        //通过SocketChannel去获得对应的管道
        ChannelPipeline pipeline = channel.pipeline();

        
        pipeline.addLast("HttpServerCodec",new HttpServerCodec());
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpObjectAggregator(1024*64));
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));//路由

        //添加自定义的助手类，给客户端浏览器渲染hello netty
        pipeline.addLast("CustomHandler",new ChatHandle());
    }
}

自定义的助手类

package websocker;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketframe;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.time.LocalDateTime;


public class ChatHandle extends SimpleChannelInboundHandler {

    //用于记录和管理所有客户端的channel
    private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketframe msg) throws Exception {
        //获取客户端所传输的载体
        String content = msg.text();
        System.out.println("接受到的消息："+content);

        //将数据刷新到客户端上
        clients.writeAndFlush(
                new TextWebSocketframe(
                        "服务器："+ LocalDateTime.now()
                        +"接收到的消息内容为："+content
                )
        );
    }

    //客户端与服务器端建立链接后，获取当前客户端所对应的channel通道
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        clients.add(ctx.channel());
    }

    //离开客户端后会自动移除channel
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户端断开，channel对应的长id："+ctx.channel().id().asLongText());
        System.out.println("客户端断开，channel对应的短id："+ctx.channel().id().asShortText());
    }
}

	
		
		
	
	
		发送消息：
		
		

		接收消息：