netty入门实现简单的echo程序

妖狐艹你老母 2024-02-19 18:38 57阅读 0赞

最近看以往在程序中编写的代码，发现有一个功能是使用socket通讯来实现的，而那个时候使用的是基于bio的阻塞io来实现的，最近在看netty,发现可以使用netty来使用nio的方式来实现，此博客记录一下netty学习的一个过程，此处实现一个简单的服务器和客户端之间的通讯。

**实现要求：**

1、服务器端监听9090端口

2、客户端连上服务端时，向服务端发送10条消息

3、使用netty自带的解码器解决tcp的粘包问题，避免接收到的数据是不完整的。（**不了解tcp粘包的可以百度一下**）

**实现：（代码的注释大致都写在了服务端这个类上）**

**一、引入netty的jar包**

<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>com.huan.netty</groupId>
    	<artifactId>netty-study</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    	<packaging>jar</packaging>
    	<name>netty-study</name>
    	<url>http://maven.apache.org</url>
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    	</properties>
    	<dependencies>
    		<dependency>
    			<groupId>junit</groupId>
    			<artifactId>junit</artifactId>
    			<version>4.10</version>
    			<scope>test</scope>
    		</dependency>
    		<dependency>
    			<groupId>io.netty</groupId>
    			<artifactId>netty-all</artifactId>
    			<version>4.1.6.Final</version>
    		</dependency>
    		<dependency>
    			<groupId>org.projectlombok</groupId>
    			<artifactId>lombok</artifactId>
    			<version>1.16.18</version>
    		</dependency>
    		<dependency>
    			<groupId>org.slf4j</groupId>
    			<artifactId>slf4j-api</artifactId>
    			<version>1.7.25</version>
    		</dependency>
    		<dependency>
    			<groupId>ch.qos.logback</groupId>
    			<artifactId>logback-classic</artifactId>
    			<version>1.1.7</version>
    		</dependency>
    		<dependency>
    			<groupId>ch.qos.logback</groupId>
    			<artifactId>logback-core</artifactId>
    			<version>1.1.7</version>
    		</dependency>
    		<dependency>
    			<groupId>ch.qos.logback</groupId>
    			<artifactId>logback-access</artifactId>
    			<version>1.1.7</version>
    		</dependency>
    	</dependencies>
    </project>

**二、netty服务端的编写**

@Slf4j
    public class EchoServer {
    
    	public static void main(String[] args) throws InterruptedException {
    		/** 此线程组用于服务单接收客户端的连接 */
    		EventLoopGroup boss = new NioEventLoopGroup(1);
    		/** 此线程组用于处理SocketChannel的读写操作 */
    		EventLoopGroup worker = new NioEventLoopGroup();
    		/** netty启动服务端的辅助类 */
    		ServerBootstrap bootstrap = new ServerBootstrap();
    		bootstrap.group(boss, worker)//
    				.channel(NioServerSocketChannel.class)// 对应的是ServerSocketChannel类
    				.option(ChannelOption.SO_BACKLOG, 128)//
    				.handler(new LoggingHandler(LogLevel.TRACE))//
    				.childHandler(new ChannelInitializer<SocketChannel>() {
    					@Override
    					protected void initChannel(SocketChannel ch) throws Exception {
    						ByteBuf delimiter = Unpooled.copiedBuffer("^^".getBytes(StandardCharsets.UTF_8));
    						// 表示客户端的数据中只要出现了^^就表示是一个完整的包，maxFrameLength这个表示如果在这个多个字节中还没有出现则表示数据有异常情况，抛出异常
    						ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
    						// 将接收到的数据转换成String的类型
    						ch.pipeline().addLast(new StringDecoder(StandardCharsets.UTF_8));
    						// 接收到的数据由自己的EchoServerHandler类进行处理
    						ch.pipeline().addLast(new EchoServerHandler());
    					}
    				});
    		// 绑定端口，同步等待成功
    		ChannelFuture future = bootstrap.bind(9090).sync();
    		log.info("server start in port:[{}]", 9090);
    		// 等待服务端链路关闭后,main线程退出
    		future.channel().closeFuture().sync();
    		// 关闭线程池资源
    		boss.shutdownGracefully();
    		worker.shutdownGracefully();
    	}
    
    	@Slf4j
    	static class EchoServerHandler extends ChannelInboundHandlerAdapter {
    		private int counter = 0;
    
    		/**
    		 * 接收到数据的时候调用
    		 */
    		@Override
    		public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    			String body = (String) msg;
    			log.info("this is: {} times receive cliemt msg: {}", ++counter, body);
    			body += "^^";
    			// 此处将数据写会到客户端，如果使用的是ctx.write方法这个只是将数据写入到缓冲区，需要再次调用ctx.flush方法
    			ctx.writeAndFlush(Unpooled.copiedBuffer(body.getBytes(StandardCharsets.UTF_8)));
    		}
    
    		/** 当发生了异常时，次方法调用 */
    		@Override
    		public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    			log.error("error:", cause);
    			ctx.close();
    		}
    	}
    
    }

**三、netty客户端代码的编写**

**客户端在服务端链接建立成功后发送了10条数据给服务端**

@Slf4j
    public class EchoClient {
    
    	public static void main(String[] args) throws InterruptedException {
    		EventLoopGroup 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 {
    						ByteBuf delimiter = Unpooled.copiedBuffer("^^".getBytes(StandardCharsets.UTF_8));
    						ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
    						ch.pipeline().addLast(new StringDecoder());
    						ch.pipeline().addLast(new EchoClientHandler());
    					}
    				});
    		ChannelFuture future = bootstrap.connect("127.0.0.1", 9090).sync();
    		log.info("client connect server.");
    		future.channel().closeFuture().sync();
    		group.shutdownGracefully();
    	}
    
    	@Slf4j
    	static class EchoClientHandler extends ChannelInboundHandlerAdapter {
    		private int counter;
    		private static final String ECHO_REQ = "hello server.服务器好啊.^^";
    
    		/**
    		 * 客户端和服务器端TCP链路建立成功后，此方法被调用
    		 */
    		@Override
    		public void channelActive(ChannelHandlerContext ctx) throws Exception {
    			for (int i = 0; i < 10; i++) {
    				ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes(StandardCharsets.UTF_8)));
    			}
    		}
    
    		/**
    		 * 接收到服务器端数据时调用
    		 */
    		@Override
    		public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    			log.info("this is :{} revice server msg:{}", ++counter, msg);
    		}
    
    		/**
    		 * 发生异常时调用
    		 */
    		@Override
    		public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    			log.error("client error:", cause);
    			ctx.close();
    		}
    	}
    }

**四、测试**  
**![aHR0cDovL2RsMi5pdGV5ZS5jb20vdXBsb2FkL2F0dGFjaG1lbnQvMDEyNy83NTcxLzQyODY0ZjBiLWIyMjctMzg0Ny04MzNkLTUwYWNmOTMzOWVhZS5wbmc][]  
服务器端和客户端都获取到了正确的数据，到此一个简单的echo工程就已经写完了。**

[aHR0cDovL2RsMi5pdGV5ZS5jb20vdXBsb2FkL2F0dGFjaG1lbnQvMDEyNy83NTcxLzQyODY0ZjBiLWIyMjctMzg0Ny04MzNkLTUwYWNmOTMzOWVhZS5wbmc]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/01330b39bc694d1bb421d50f3bf93f10.png