Netty 的基本简单实例【服务端-客户端通信】

Netty是建立在NIO基础之上，Netty在NIO之上又提供了更高层次的抽象。

在Netty里面，Accept连接可以使用单独的线程池去处理，读写操作又是另外的线程池来处理。

Accept连接和读写操作也可以使用同一个线程池来进行处理。而请求处理逻辑既可以使用单独的线程池进行处理，也可以跟放在读写线程一块处理。线程池中的每一个线程都是NIO线程。用户可以根据实际情况进行组装，构造出满足系统需求的并发模型。

Netty提供了内置的常用编解码器，包括行编解码器［一行一个请求］，前缀长度编解码器［前N个字节定义请求的字节长度］，可重放解码器［记录半包消息的状态］，HTTP编解码器，WebSocket消息编解码器等等

Netty提供了一些列生命周期回调接口，当一个完整的请求到达时，当一个连接关闭时，当一个连接建立时，用户都会收到回调事件，然后进行逻辑处理。

Netty可以同时管理多个端口，可以使用NIO客户端模型，这些对于RPC服务是很有必要的。

Netty除了可以处理TCP Socket之外，还可以处理UDP Socket。

在消息读写过程中，需要大量使用ByteBuffer，Netty对ByteBuffer在性能和使用的便捷性上都进行了优化和抽象。

服务端：

package com.kenson.netty.server;

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.Channel;

import io.netty.channel.ChannelOption;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.nio.NioServerSocketChannel;

public class NettyServer {

    /**

     * 端口

     */

    private int port;

    public NettyServer(int port) {

        this.port = port;

    }

    public void run() {

        //EventLoopGroup是用来处理IO操作的多线程事件循环器

        //负责接收客户端连接线程

        EventLoopGroup bossGroup = new NioEventLoopGroup();

        //负责处理客户端i/o事件、task任务、监听任务组

        EventLoopGroup workerGroup = new NioEventLoopGroup();

        //启动 NIO 服务的辅助启动类

        ServerBootstrap bootstrap = new ServerBootstrap();

        bootstrap.group(bossGroup, workerGroup);

        //配置 Channel

        bootstrap.channel(NioServerSocketChannel.class);

        bootstrap.childHandler(new ServerIniterHandler());

        //BACKLOG用于构造服务端套接字ServerSocket对象，

        // 标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度

        bootstrap.option(ChannelOption.SO_BACKLOG, 1024);

        //是否启用心跳保活机制

        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);

        try {

            //绑定服务端口监听

            Channel channel = bootstrap.bind(port).sync().channel();

            System.out.println("server run in port " + port);

            //服务器关闭监听

            /*channel.closeFuture().sync()实际是如何工作:

            channel.closeFuture()不做任何操作，只是简单的返回channel对象中的closeFuture对象，对于每个Channel对象，都会有唯一的一个CloseFuture，用来表示关闭的Future，

            所有执行channel.closeFuture().sync()就是执行的CloseFuturn的sync方法，从上面的解释可以知道，这步是会将当前线程阻塞在CloseFuture上*/

            channel.closeFuture().sync();

        } catch (InterruptedException e) {

            e.printStackTrace();

        } finally {

            //关闭事件流组

            bossGroup.shutdownGracefully();

            workerGroup.shutdownGracefully();

        }

    }

    public static void main(String[] args) {

        new NettyServer(8899).run();

    }

}

服务端业务逻辑处理：

package com.kenson.netty.server;

import io.netty.channel.Channel;

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.util.concurrent.GlobalEventExecutor;

public class ServerHandler extends SimpleChannelInboundHandler<String> {

    /**

     * 所有的活动用户

     */

    public static final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**

     * 读取消息通道

     *

     * @param context

     * @param s

     * @throws Exception

     */

    @Override

    protected void channelRead0(ChannelHandlerContext context, String s)

            throws Exception {

        Channel channel = context.channel();

        //当有用户发送消息的时候，对其他的用户发送消息

        for (Channel ch : group) {

            if (ch == channel) {

                ch.writeAndFlush("[you]: " + s + "\n");

            } else {

                ch.writeAndFlush("[" + channel.remoteAddress() + "]: " + s + "\n");

            }

        }

        System.out.println("[" + channel.remoteAddress() + "]: " + s + "\n");

    }

    /**

     * 处理新加的消息通道

     *

     * @param ctx

     * @throws Exception

     */

    @Override

    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {

        Channel channel = ctx.channel();

        for (Channel ch : group) {

            if (ch == channel) {

                ch.writeAndFlush("[" + channel.remoteAddress() + "] coming");

            }

        }

        group.add(channel);

    }

    /**

     * 处理退出消息通道

     *

     * @param ctx

     * @throws Exception

     */

    @Override

    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {

        Channel channel = ctx.channel();

        for (Channel ch : group) {

            if (ch == channel) {

                ch.writeAndFlush("[" + channel.remoteAddress() + "] leaving");

            }

        }

        group.remove(channel);

    }

    /**

     * 在建立连接时发送消息

     *

     * @param ctx

     * @throws Exception

     */

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        Channel channel = ctx.channel();

        boolean active = channel.isActive();

        if (active) {

            System.out.println("[" + channel.remoteAddress() + "] is online");

        } else {

            System.out.println("[" + channel.remoteAddress() + "] is offline");

        }

        ctx.writeAndFlush("[server]: welcome");

    }

    /**

     * 退出时发送消息

     *

     * @param ctx

     * @throws Exception

     */

    @Override

    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

        Channel channel = ctx.channel();

        if (!channel.isActive()) {

            System.out.println("[" + channel.remoteAddress() + "] is offline");

        } else {

            System.out.println("[" + channel.remoteAddress() + "] is online");

        }

    }

    /**

     * 异常捕获

     *

     * @param ctx

     * @param e

     * @throws Exception

     */

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {

        Channel channel = ctx.channel();

        System.out.println("[" + channel.remoteAddress() + "] leave the room");

        ctx.close().sync();

    }

}

服务端处理器注册：

package com.kenson.netty.server;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelPipeline;

import io.netty.channel.socket.SocketChannel;

import io.netty.handler.codec.string.StringDecoder;

import io.netty.handler.codec.string.StringEncoder;

public class ServerIniterHandler extends ChannelInitializer<SocketChannel> {

    @Override

    protected void initChannel(SocketChannel socketChannel) throws Exception {

        //管道注册handler

        ChannelPipeline pipeline = socketChannel.pipeline();

        //编码通道处理

        pipeline.addLast("decode", new StringDecoder());

        //转码通道处理

        pipeline.addLast("encode", new StringEncoder());

        //聊天服务通道处理

        pipeline.addLast("chat", new ServerHandler());

    }

}

客户端：

package com.kenson.netty.client;

import io.netty.bootstrap.Bootstrap;

import io.netty.channel.Channel;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.nio.NioSocketChannel;

import org.apache.commons.lang3.StringUtils;

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStreamReader;

public class NettyClient {

    private String ip;

    private int port;

    private boolean stop = false;

    public NettyClient(String ip, int port) {

        this.ip = ip;

        this.port = port;

    }

    public void run() throws IOException {

        //设置一个多线程循环器

        EventLoopGroup workerGroup = new NioEventLoopGroup();

        //启动附注类

        Bootstrap bootstrap = new Bootstrap();

        bootstrap.group(workerGroup);

        //指定所使用的NIO传输channel

        bootstrap.channel(NioSocketChannel.class);

        //指定客户端初始化处理

        bootstrap.handler(new ClientIniterHandler());

        try {

            //连接服务

            Channel channel = bootstrap.connect(ip, port).sync().channel();

            while (true) {

                //向服务端发送内容

                BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));

                String content = reader.readLine();

                if (StringUtils.isNotEmpty(content)) {

                    if (StringUtils.equalsIgnoreCase(content, "q")) {

                        System.exit(1);

                    }

                    channel.writeAndFlush(content);

                }

            }

        } catch (InterruptedException e) {

            e.printStackTrace();

            System.exit(1);

        } finally {

            workerGroup.shutdownGracefully();

        }

    }

    public static void main(String[] args) throws Exception {

        new NettyClient("127.0.0.1", 8899).run();

    }

}

客户端逻辑处理：

package com.kenson.netty.client;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.SimpleChannelInboundHandler;

public class ClientHandler extends SimpleChannelInboundHandler<String> {

    @Override

    protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {

        //打印服务端的发送数据

        System.out.println(s);

    }

}

客户端处理器注册：

package com.kenson.netty.client;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelPipeline;

import io.netty.channel.socket.SocketChannel;

import io.netty.handler.codec.http.HttpClientCodec;

import io.netty.handler.codec.string.StringDecoder;

import io.netty.handler.codec.string.StringEncoder;

public class ClientIniterHandler extends ChannelInitializer<SocketChannel> {

    @Override

    protected void initChannel(SocketChannel socketChannel) throws Exception {

        //注册管道

        ChannelPipeline pipeline = socketChannel.pipeline();

        pipeline.addLast("decoder", new StringDecoder());

        pipeline.addLast("encoder", new StringEncoder());

        pipeline.addLast("http", new HttpClientCodec());

        pipeline.addLast("chat", new ClientHandler());

    }

}

测试时先启动服务端，再启动客户端。。。

声明：本文借鉴网络资源，来自：https://www.cnblogs.com/kingsonfu/p/8635064.html

如果侵犯您的利益，请联系本人删除

巴特西

Netty 的基本简单实例【服务端-客户端通信】

最新文章

热门文章