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的核数*2EventLoopGroup bossGroup = new NioEventLoopGroup();//真正的和客户端进行业务处理，含有的子线程有多少个呢，NioEventLoop个数，默认是cpu的核数*2EventLoopGroup workerGroup = new NioEventLoopGroup();try {//创建服务端的启动对象，设置参数ServerBootstrap bootstrap = new ServerBootstrap();//设置两个线程组boosGroup和workerGroupbootstrap.group(bossGroup, workerGroup)//设置服务端通道实现类型.channel(NioServerSocketChannel.class) //使用nioserversocketchannel作为通道实现//设置线程队列得到连接个数.option(ChannelOption.SO_BACKLOG, 128) //设置线程队列等待连接个数//设置保持活动连接状态.childOption(ChannelOption.SO_KEEPALIVE, true)//使用匿名内部类的形式初始化通道对象.childHandler(new ChannelInitializer<SocketChannel>() {//这里会进行客户端业务处理//通过一个特殊的ChannelInboundHandler 来初始化注册到EventLoop的Channel@Overrideprotected void initChannel(SocketChannel sc) throws Exception {//给pipeline管道设置处理器ChannelPipeline pipeline = sc.pipeline();//添加一个基于行的解码器pipeline.addLast(new LineBasedFrameDecoder(2048));pipeline.addLast(new StringDecoder());pipeline.addLast(new StringEncoder());//这里加入了自定义的handlerpipeline.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与服务端连接成功就会执行这个方法@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 获取到当前与服务器连接成功的channelChannel 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断开与服务端的连接就会执行这个方法@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {// 获取到当前要断开连接的ChannelChannel 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给当前的服务端发送了消息，那么就会触发该方法的执行*/@Overrideprotected 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，而是使用bootstrapBootstrap bootstrap = new Bootstrap();//设置线程组bootstrap.group(group).channel(NioSocketChannel.class) //设置客户端的通道实现类型.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected 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);// 将输入的内容写入到Channelwhile (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();}
}