Netty编码示例（RPC & WbeSocket & Tomcat）

public class Constant {
    public static final String protocolName = "cityService#name#";
}

public class CityServiceImpl implements CityService {
    private  int count = 0;
    @Override
    public String getName(String mes) {
        System.out.println("收到客户端消息=" + mes);
        //根据mes 返回不同的结果
        if (mes != null) {
            return "你好客户端, 我已经收到你的消息 [" + mes + "] 第" + (++count) + " 次";
        } else {
            return "你好客户端, 我已经收到你的消息 ";
        }
    }
}

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    private CityService cityService = null;

    public NettyServerHandler(CityService cityService ){
        this.cityService = cityService;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //获取客户端发送的消息，并调用服务
        System.out.println("msg=" + msg);
        //客户端在调用服务器的api时，我们需要定义一个协议
        //比如我们要求 每次发消息是都必须以某个字符串开头 "cityService#name##你好"
        if (msg.toString().startsWith(Constant.protocolName)) {
            String result = cityService.getName(msg.toString().substring(Constant.protocolName.length()));
            ctx.writeAndFlush(result);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

public class NettyServer {
    public static void startServer(String hostName, int port) {
        startServer0(hostName, port);
    }

    //编写一个方法，完成对NettyServer的初始化和启动
    private static void startServer0(String hostname, int port) {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            CityService cityService = new CityServiceImpl();
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                                      @Override
                                      protected void initChannel(SocketChannel ch) throws Exception {
                                          ChannelPipeline pipeline = ch.pipeline();
                                          pipeline.addLast(new StringDecoder());
                                          pipeline.addLast(new StringEncoder());
                                          pipeline.addLast(new NettyServerHandler(cityService)); //业务处理器

                                      }
                                  }
                    );
            ChannelFuture channelFuture = serverBootstrap.bind(hostname, port).sync();
            System.out.println("server is ready");
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

public class ServerBootstrap {
    public static void main(String[] args) {
        NettyServer.startServer("127.0.0.1", 7000);
    }
}

public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {

    private ChannelHandlerContext context;//上下文
    private String result; //返回的结果
    private String para; //客户端调用方法时，传入的参数

    //与服务器的连接创建后，就会被调用, 这个方法是第一个被调用(1)/(2)
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channelActive call,thread="+Thread.currentThread().getName());
        context = ctx; //因为我们在其它方法会使用到 ctx
    }

    //收到服务器的数据后，调用方法 (4)
    //
    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("channelRead call,thread="+Thread.currentThread().getName());
        result = msg.toString();
        notify(); //唤醒等待的线程
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

    //被代理对象调用, 发送数据给服务器，-> wait -> 等待被唤醒(channelRead) -> 返回结果 (3)-》5
    @Override
    public synchronized Object call() throws Exception {
        System.out.println("before send,thread="+Thread.currentThread().getName());
        context.writeAndFlush(para);
        //进行wait
        wait(); //等待channelRead 方法获取到服务器的结果后，唤醒
        System.out.println("after send,thread="+Thread.currentThread().getName());
        return result; //服务方返回的结果
    }

    void setPara(String para) {
        System.out.println("setPara,thread="+Thread.currentThread().getName());
        this.para = para;
    }
}

public class NettyClient {
    //创建线程池
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    private static NettyClientHandler nettyClientHandler;
    private int count = 0;
    //编写方法使用代理模式，获取一个代理对象
    public Object getBean(final Class<?> serivceClass, final String providerName) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class<?>[]{serivceClass}, (proxy, method, args) -> {
                    System.out.println("(proxy, method, args) 进入...." + (++count) + " 次");
                    //{}  部分的代码，客户端每调用一次 hello, 就会进入到该代码
                    if (nettyClientHandler == null) {
                        initNettyClientHandler();
                    }
                    System.out.println("method:"+method.getName());
                    //设置要发给服务器端的信息
                    //providerName 协议头 args[0] 就是客户端调用api getName(???), 参数
                    nettyClientHandler.setPara(providerName + args[0]);                    return executor.submit(nettyClientHandler).get();
                });
    }

    //初始化客户端
    private static void initNettyClientHandler() {
        nettyClientHandler = new NettyClientHandler();
        //创建EventLoopGroup
        NioEventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(
                        new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ChannelPipeline pipeline = ch.pipeline();
                                pipeline.addLast(new StringDecoder());
                                pipeline.addLast(new StringEncoder());
                                pipeline.addLast(nettyClientHandler);
                            }
                        }
                );
        try {
            bootstrap.connect("127.0.0.1", 7000).sync();
        } catch (Exception e) {
            e.printStackTrace();
        }
        //重要！不要关闭
       // group.shutdownGracefully();
    }
}

public class ClientBootstrap {
    //这里定义协议头
    public static void main(String[] args) throws Exception {
        //创建一个消费者
        NettyClient customer = new NettyClient();
        //创建代理对象
        CityService service = (CityService) customer.getBean(CityService.class, Constant.protocolName);
        for(int i=0;i<10;i++){
            //通过代理对象调用服务提供者的方法(服务)
            String res = service.getName("hello bejing-"+i);
            System.out.println("调用的结果 res= " + res);
            Thread.sleep(5 * 1000);
        }
    }
}

msg=cityService#name#hello bejing-0
收到客户端消息=hello bejing-0
msg=cityService#name#hello bejing-1
收到客户端消息=hello bejing-1
msg=cityService#name#hello bejing-2
收到客户端消息=hello bejing-2
msg=cityService#name#hello bejing-3
收到客户端消息=hello bejing-3

(proxy, method, args) 进入....1 次
method:getName
channelActive call,thread=nioEventLoopGroup-2-1
setPara,thread=main
before send,thread=pool-1-thread-1
channelRead call,thread=nioEventLoopGroup-2-1
after send,thread=pool-1-thread-1
调用的结果 res= 你好客户端, 我已经收到你的消息 [hello bejing-0] 第1 次
(proxy, method, args) 进入....2 次
method:getName
setPara,thread=main
before send,thread=pool-1-thread-2
channelRead call,thread=nioEventLoopGroup-2-1
after send,thread=pool-1-thread-2
调用的结果 res= 你好客户端, 我已经收到你的消息 [hello bejing-1] 第2 次
(proxy, method, args) 进入....3 次
method:getName
setPara,thread=main
before send,thread=pool-1-thread-3
channelRead call,thread=nioEventLoopGroup-2-1
after send,thread=pool-1-thread-3
调用的结果 res= 你好客户端, 我已经收到你的消息 [hello bejing-2] 第3 次

public class MyWebSocketServer {
    int port;
    public MyWebSocketServer(int port) {
        this.port = port;
    }
    public void start() {
        ServerBootstrap bootstrap = new ServerBootstrap();
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup work = new NioEventLoopGroup();
        try {
            bootstrap.group(boss, work)
                    .handler(new LoggingHandler(LogLevel.DEBUG))
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new WebSocketServerInitializer());
            ChannelFuture f = bootstrap.bind(new InetSocketAddress(port)).sync();
            System.out.println("server started . port : " + port);
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            boss.shutdownGracefully();
            work.shutdownGracefully();
        }
    }

    public void pushMsg(){
        //模拟异步发送推送消息
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(1);
        scheduledThreadPool.scheduleWithFixedDelay(() -> {
            TextWebSocketFrame tws = new TextWebSocketFrame("server push message,this time："+ LocalDateTime.now());
            // 群发
            ChannelSupervise.send2All(tws);
        }, 0,1, TimeUnit.SECONDS);
    }

    public static void main(String[] args) {
        MyWebSocketServer server = new MyWebSocketServer(8081);// 8081为启动端口
//        server.pushMsg();
        server.start();
    }
} 

public class ChannelSupervise {
    /**
     * ChannelGroup是netty提供用于管理web于服务器建立的通道channel的，
     * 其本质是一个高度封装的set集合，在服务器广播消息时，
     * 可以直接通过它的writeAndFlush将消息发送给集合中的所有通道中去
     */
    private static ChannelGroup GlobalGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    /**
     * ChannelMap维护的是channelID和channel的对应关系，用于向指定channel发送消息
     */
    private static ConcurrentMap<String, ChannelId> ChannelMap = new ConcurrentHashMap<>();

    public static void addChannel(Channel channel) {
        GlobalGroup.add(channel);
        ChannelMap.put(channel.id().asShortText(), channel.id());
    }

    public static void removeChannel(Channel channel) {
        GlobalGroup.remove(channel);
        ChannelMap.remove(channel.id().asShortText());
    }

    //找到某个channel来发送消息
    public static Channel findChannel(String id) {
        return GlobalGroup.find(ChannelMap.get(id));
    }

    public static void send2All(TextWebSocketFrame tws) {
        GlobalGroup.writeAndFlush(tws);
    }
} 

public abstract class WebSocketServerHandshaker {
　　....
　　//WebSocket相关的编解码
　　public final ChannelFuture handshake(Channel channel, FullHttpRequest req, HttpHeaders responseHeaders, final ChannelPromise promise) {
　　　　if (logger.isDebugEnabled()) {
　　　　　　logger.debug("{} WebSocket version {} server handshake", channel,
　　　　　　version());
　　　　}
　　　　//构造握⼿响应
　　　　FullHttpResponse response = newHandshakeResponse(req, responseHeaders);
　　　　//下⾯将channelpipeline中的HttpObjectAggregator和HttpContentCompressor移除，并且添加WebSocket编解码器newWebSocketEncoder和newWebsocketDecoder
　　　　ChannelPipeline p = channel.pipeline();
　　　　if (p.get(HttpObjectAggregator.class) != null) {
　　　　　　p.remove(HttpObjectAggregator.class);
　　　　}
　　　　if (p.get(HttpContentCompressor.class) != null) {
　　　　　　p.remove(HttpContentCompressor.class);
　　　　}
　　　　ChannelHandlerContext ctx = p.context(HttpRequestDecoder.class);
　　　　final String encoderName;
　　　　if (ctx == null) {
　　　　　　// this means the user use an HttpServerCodec
　　　　　　ctx = p.context(HttpServerCodec.class);
　　　　　　if (ctx == null) {
　　　　　　　　promise.setFailure(new IllegalStateException("No HttpDecoder and no HttpServerCodec in the pipeline"));
　　　　　　　　return promise;
　　　　　　}
　　　　　　p.addBefore(ctx.name(), "wsencoder", newWebSocketEncoder());
　　　　　　p.addBefore(ctx.name(), "wsdecoder", newWebsocketDecoder());
　　　　　　encoderName = ctx.name();
　　　　} else {
　　　　　　p.replace(ctx.name(), "wsdecoder", newWebsocketDecoder());
　　　　　　encoderName = p.context(HttpResponseEncoder.class).name();
　　　　　　p.addBefore(encoderName, "wsencoder", newWebSocketEncoder());
　　　　}
　　　　//将response消息返回给客户端
　　　　channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
　　　　　　@Override
　　　　　　public void operationComplete(ChannelFuture future) throws Exception {
　　　　　　　　if (future.isSuccess()) {
　　　　　　　　　　ChannelPipeline p = future.channel().pipeline();
　　　　　　　　　　p.remove(encoderName);
　　　　　　　　　　promise.setSuccess();
　　　　　　　　} else {
　　　　　　　　　　promise.setFailure(future.cause());
　　　　　　　　} 
　　　　　　}
　　　　});
　　　　return promise;
　　}
...

public class HttpHelloWorldServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    public void initChannel(SocketChannel ch) {
        ChannelPipeline p = ch.pipeline();
        /**
         * 或者使用HttpRequestDecoder & HttpResponseEncoder
         * HttpRequestDecoder 即把 ByteBuf 解码到 HttpRequest 和 HttpContent。
         * HttpResponseEncoder 即把 HttpResponse 或 HttpContent 编码到 ByteBuf。
         * HttpServerCodec 即 HttpRequestDecoder 和 HttpResponseEncoder 的结合。
         */
        p.addLast(new HttpServerCodec());
        /**
         * 在处理POST消息体时需要加上
         */
        p.addLast(new HttpObjectAggregator(1024*1024));
        //这个方法的作用是: http 100-continue用于客户端在发送POST数据给服务器前，
        // 征询服务器情况，看服务器是否处理POST的数据，
        // 如果不处理，客户端则不上传POST数据，如果处理，则POST上传数据。
        // 在现实应用中，通过在POST大数据时，才会使用100-continue协议
        //
        p.addLast(new HttpServerExpectContinueHandler());
        p.addLast(new HttpHelloWorldServerHandler());
    }
}

public class HttpHelloWorldServerHandler extends SimpleChannelInboundHandler<HttpObject> {
    private HttpHeaders headers;
    private HttpRequest request;
    private FullHttpRequest fullRequest;
    private static final AsciiString CONTENT_TYPE = AsciiString.cached("Content-Type");
    private static final AsciiString CONTENT_LENGTH = AsciiString.cached("Content-Length");
    private static final AsciiString CONNECTION = AsciiString.cached("Connection");
    private static final AsciiString KEEP_ALIVE = AsciiString.cached("keep-alive");

    @Override
    public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
        User user = new User();
        user.setDate(new Date());
        if (msg instanceof HttpRequest){
            request = (HttpRequest) msg;
            headers = request.headers();
            String uri = request.uri();
            System.out.println("http uri: "+ uri);
            HttpMethod method = request.method();
            //localhost:8888/user?name=北京
            if (method.equals(HttpMethod.GET)){
                QueryStringDecoder queryDecoder = new QueryStringDecoder(uri, CharsetUtil.UTF_8);
                Map<String, List<String>> uriAttributes = queryDecoder.parameters();
                for (Map.Entry<String, List<String>> attr : uriAttributes.entrySet()) {
                    for (String attrVal : attr.getValue()) {
                        System.out.println(attr.getKey() + "=" + attrVal);
                        user.setUserName(attrVal);
                    }
                }
                user.setMethod("get");
            }else if (method.equals(HttpMethod.POST)){
                //POST请求,由于你需要从消息体中获取数据,因此有必要把msg转换成FullHttpRequest
                //即 FullHttpRequest 包含了 HttpRequest 和 FullHttpMessage，是一个 HTTP 请求的完全体。
                fullRequest = (FullHttpRequest)msg;
                //根据不同的Content_Type处理body数据
                String result = dealWithContentType();
                user.setMethod("post");
                user.setUserName(result);

            }
            byte[] content =JSON.toJSONBytes(user);
            FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(content));
            response.headers().set(CONTENT_TYPE, "text/plain");
            response.headers().setInt(CONTENT_LENGTH, response.content().readableBytes());
            boolean keepAlive = HttpUtil.isKeepAlive(request);
            if (!keepAlive) {
                ctx.write(response).addListener(ChannelFutureListener.CLOSE);
            } else {
                response.headers().set(CONNECTION, KEEP_ALIVE);
                ctx.write(response);
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    /**
     * 简单处理常用几种 Content-Type 的 POST 内容（可自行扩展）
     * @throws Exception
     */
    private String dealWithContentType() throws Exception{
        String result = "";
        String contentType = getContentType();
        //可以使用HttpJsonDecoder
        if(contentType.equals("application/json")){
            String jsonStr = fullRequest.content().toString(CharsetUtil.UTF_8);
            JSONObject obj = JSON.parseObject(jsonStr);
            for(Map.Entry<String, Object> item : obj.entrySet()){
                System.out.println(item.getKey()+"="+item.getValue().toString());
                result = item.getValue().toString();
            }

        }else if(contentType.equals("application/x-www-form-urlencoded")){
            //方式一：使用 QueryStringDecoder
            String jsonStr = fullRequest.content().toString(CharsetUtil.UTF_8);
            QueryStringDecoder queryDecoder = new QueryStringDecoder(jsonStr, false);
            Map<String, List<String>> uriAttributes = queryDecoder.parameters();
            for (Map.Entry<String, List<String>> attr : uriAttributes.entrySet()) {
                for (String attrVal : attr.getValue()) {
                    System.out.println(attr.getKey() + "=" + attrVal);
                }
            }

        }else if(contentType.equals("multipart/form-data")){
            //TODO 用于文件上传
        }else{
            //do nothing...
        }
        return result;
    }
    private String getContentType(){
        String typeStr = headers.get("Content-Type").toString();
        String[] list = typeStr.split(";");
        return list[0];
    }

}

public final class HttpHelloWorldServer {
    private static final Logger logger = LoggerFactory.getLogger(HttpHelloWorldServer.class);

    static final int PORT = 8888;

    public static void main(String[] args) throws Exception {
        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.option(ChannelOption.SO_BACKLOG, 1024);
            b.childOption(ChannelOption.TCP_NODELAY,true);
            b.childOption(ChannelOption.SO_KEEPALIVE,true);
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new HttpHelloWorldServerInitializer());

            Channel ch = b.bind(PORT).sync().channel();
            logger.info("Netty http server listening on port "+ PORT);
            ch.closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}