netty学习（1）:1个客户端与服务器通信

1. 新建maven工程，添加netty依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>nettyTest</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

    <dependencies>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.34.Final</version>
        </dependency>
    </dependencies>
</project>

2. 新建netty服务器

package server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.concurrent.ConcurrentHashMap;

public class NettyService {
    /**
     * 1.创建BossGroup线程组，处理网络连接事件
     * 2.创建workerGroup线程组 处理网络读写事件
     * 3.创建服务端启动助手，serverBootStrap
     * 4.设置服务端通道实现方式为NIO
     * 5.设置服务端options
     * 6.创建通道初始化对象
     * 8.向pipeline中添加自定义业务初处理逻辑handler
     * 9.启动服务端并绑定端口，将异步改为同步
     * 10.关闭通道和连接池
     */

    //监听的端口号
    private final int port;
    public NettyService(int port) {
        this.port = port;
    }

    public void createNettyService() {
        //创建两个线程组 boosGroup、workerGroup,这两个都是无限循环,boos是负责进行连接请求的接收accept事件，而worker则是负责业务处理
        //只是处理连接请求accept，含有的子线程有多少个呢，NioEventLoop个数，默认是cpu的核数*2
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        //真正的和客户端进行业务处理，含有的子线程有多少个呢，NioEventLoop个数，默认是cpu的核数*2
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //创建服务端的启动对象，设置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            //设置两个线程组boosGroup和workerGroup
            bootstrap.group(bossGroup, workerGroup)
                    //设置服务端通道实现类型
                    .channel(NioServerSocketChannel.class) //使用nioserversocketchannel作为通道实现
                    //设置线程队列得到连接个数
                    .option(ChannelOption.SO_BACKLOG, 128) //设置线程队列等待连接个数
                    //设置保持活动连接状态
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //使用匿名内部类的形式初始化通道对象
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        //这里会进行客户端业务处理
                        //通过一个特殊的ChannelInboundHandler 来初始化注册到EventLoop的Channel
                        @Override
                        protected void initChannel(SocketChannel sc) throws Exception {
                            //给pipeline管道设置处理器
                            ChannelPipeline pipeline = sc.pipeline();
                            //添加一个基于行的解码器
                            pipeline.addLast(new LineBasedFrameDecoder(2048));
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            //这里加入了自定义的handler
                            pipeline.addLast(new NettyServiceHandler());
                        }
                    });//给workerGroup的EventLoop对应的管道设置处理器
            //绑定端口号，启动服务端
            ChannelFuture channelFuture = bootstrap.bind(port).sync();
            System.out.println("netty服务端已经准备就绪,端口" + port);
            //对关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

自定义handler

package server;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * 自定义处理Handler
 */
public class NettyServiceHandler extends SimpleChannelInboundHandler<String> {
    // 创建一个ChannelGroup，其是一个线程安全的集合，其中存放着与当前服务器相连接的所有Active状态的Channel
    // GlobalEventExecutor是一个单例、单线程的EventExecutor，是为了保证对当前group中的所有Channel的处理
    // 线程是同一个线程
    //private static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    // 只要有客户端Channel与服务端连接成功就会执行这个方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 获取到当前与服务器连接成功的channel
        Channel channel = ctx.channel();
        System.out.println(channel.remoteAddress() + "---上线");
        //group.writeAndFlush(channel.remoteAddress() + "---上线\n");
        // 将当前channel添加到group中
        //group.add(channel);

//        NettyService.remoteAddressMap.put(channel.remoteAddress().toString(), ctx);
    }

    // 只要有客户端Channel断开与服务端的连接就会执行这个方法
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 获取到当前要断开连接的Channel
        Channel channel = ctx.channel();
        System.out.println(channel.remoteAddress() + "------下线");
        //group.writeAndFlush(channel.remoteAddress() + "下线，当前在线人数：" + group.size() + "\n");

        // group中存放的都是Active状态的Channel，一旦某Channel的状态不再是Active，
        // group会自动将其从集合中踢出，所以，下面的语句不用写
        // remove()方法的应用场景是，将一个Active状态的channel移出group时使用
        // group.remove(channel);
    }

    /**
     * channelRead，这个方法当有数据读写的时候，会触发，可以读取客户端的消息
     * 只要有客户端Channel给当前的服务端发送了消息，那么就会触发该方法的执行
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        Channel channel = ctx.channel();
        System.out.println("netty客户端" + channel.remoteAddress() + "发送过来的消息:" + msg);
        channel.writeAndFlush("自己发的消息：" + msg + "\n");
        
//        String[] split = msg.split(":");
//        if (split.length >= 2) { //指定客户端时发送给指定客户
//            System.out.println("发送给客户:" + split[0]);
//            if (NettyService.userIdMap.get(split[0]) != null) {
//                NettyService.userIdMap.get(split[0]).writeAndFlush(channel.remoteAddress() + "：" + split[1] + "\n");
//            } else {
//                NettyService.userIdMap.put(split[0], ctx);  //第一次发送消息时注册客户端的channel
//            }
//        } else { //否则发送给所有的客户端
//            System.out.println("广播");
//            // 遍历channelGroup，从而区分“我”和“别人”发出的消息，如果消息是自己发出的就显示“我”
//            group.forEach(ch -> { // JDK8 提供的lambda表达式
//                if (ch != channel) {
//                    ch.writeAndFlush(channel.remoteAddress() + "：" + msg + "\n");
//                } else {
//                    channel.writeAndFlush("自己发的消息：" + msg + "\n");
//                }
//            });
//        }
    }
    /**
     * channelReadComplete，数据读取完毕之后，需要做的业务操作，回消息
     */
//    public void channelReadComplete(ChannelHandlerContext channelHandlerContext) throws Exception {
//        //消息出站
//        System.out.println("Netty服务端读取消息完毕");
//        channelHandlerContext.writeAndFlush(Unpooled.copiedBuffer("--over", CharsetUtil.UTF_8));
//    }

    /**
     * exceptionCaught，发生异常的handler
     */
    public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception {
        throwable.printStackTrace();
        channelHandlerContext.close();
    }
}

3. 新建客户端

package client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;

public class NettyClient {
    /**
     * 1.创建线程组
     * 2.设置线程组启动助手 bootstrap
     * 3.设置客户端通道为NIO
     * 4.创建通道初始化对象
     * 5.向pipeline中添加自定义业务处理的handler
     * 6.启动客户端，等待链接服务端，同时将异步改为同步
     * 8.关闭通道和连接池
     */
    private final String host;
    private final int port;
    public NettyClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void createNettyClient() {
        //客户端跟服务端有些许区别，第一个就是不需要两个loopgroup了，不需要使用boosgroup了，所以只需要新建一个loopgroup来处理业务就行了
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //第二点就是不是使用serverBootstrap，而是使用bootstrap
            Bootstrap bootstrap = new Bootstrap();
            //设置线程组
            bootstrap.group(group)
                    .channel(NioSocketChannel.class) //设置客户端的通道实现类型
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new LineBasedFrameDecoder(2048));
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(new NettyClientHandler()); //加入自己的处理器
                        }
                    });
            //启动客户端
            ChannelFuture future = bootstrap.connect(host, port).sync();
            System.out.println("netty客户端1准备就绪");
            // 获取键盘输入
            InputStreamReader is = new InputStreamReader(System.in, StandardCharsets.UTF_8);
            BufferedReader br = new BufferedReader(is);
            // 将输入的内容写入到Channel
            while (true) {
                //br.readLine()中执行fill()方法获取输入数据，获取不到时会发生阻塞，直到获取到数据为止
                future.channel().writeAndFlush(br.readLine() + "\r\n");
            }

            //监听通道关闭的状态事件
            //channel.closeFuture().sync();
        } catch (InterruptedException | IOException e) {
            e.printStackTrace();
        } finally {
            //关闭通道和连接池
            group.shutdownGracefully();
        }
    }
}

自定义handler

package client;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * 自定义处理Handler
 */
public class NettyClientHandler extends SimpleChannelInboundHandler<String> {
    /**
     * channelActive，当通道就绪，就会触发该方法
     */
//    public void channelActive(ChannelHandlerContext channelHandlerContext) throws Exception {
//        //发送消息到服务端
//        channelHandlerContext.writeAndFlush(Unpooled.copiedBuffer("netty客户端1启动.", CharsetUtil.UTF_8));
//    }

//    public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception {
//
//    }
    /**
     * channelRead，当通道有读取事件时，会触发
     */
    public void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {//接收服务端发送过来的消息
        System.out.println("收到服务端消息：" + channelHandlerContext.channel().remoteAddress() + "的消息：" + msg);
    }

    /**
     * exceptionCaught，当有异常的就会触发
     */
    public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception {
        channelHandlerContext.close();
    }
}

4. 测试

启动服务器

package test;

import server.NettyService;

public class Service {
    public static void main(String[] args) {
        new NettyService(6666).createNettyService();
    }
}

启动客户端

package test;

import client.NettyClient;

public class Client1 {
    public static void main(String[] args) {
        new NettyClient("127.0.0.1",6666).createNettyClient();
    }
}