Netty简单实现一个回声服务器

Netty简单实现一个回声服务器

预备知识点及思路：

• Netty 组件的基本使用
• 网络通信信道是全双工的：
  • 需要注意的是使用底层的Sokcet 的时候用的是流，更容易被理解为对于通信传输的网络通道是单向的，写的时候不能传数据，传数据的时候不能写数据而实际上是网络信道是可以同时进行通信传输的，而Channel 能同时进行读 *** 作和写 *** 作其实是更贴近底层的一种模拟。
  • 对于单线程来说，数据的处理是顺序的，这也就造成了又一层的误解，如果只按单线程顺序的思想就无法理解同时进行写数据和读数据的 *** 作，这里就需要理解多线程的思想，一个线程向Channel 中写数据，一个线程从Channel 中读数据。
• 在Idea 的控制台进行数据的读取是需要一个线程专门去维护的，否则，按照顺序结果只能完成简单的一写一读的过程（从需求上看我们貌似是这样的，但是这是理解上的差距，所以这里一定要强调），而不能实现实时的多写多读的过程。通过一个线程去读控制台收入的用户输入的数据，并发送给服务，再由netty 的读线程读取channel 中的数据答应输出，这个过程程序的实现逻辑上是由先后顺序的，但是从网络的原理上，服务器发数据和客户端向服务器写数据是不存在必须的先后关系的。
• socket 是TCP协议的传输目的地和源点，所以说其实是可以实现自己定制应用层协议（类似于HTTP协议或者SMTP协议）
• 运输层协议的传输单位是字节，所以在这里的 *** 作都是通过ByteBuf 进行的

实现：

EcoServer

package practice1.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;


public class EcoServer {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                       ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                           @Override
                           public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                               System.out.println("连接建立成功："+ctx.channel());
                               ByteBuf message = msg instanceof ByteBuf ? (ByteBuf) msg: null;

                               
                               System.out.println("From:" + ctx.channel());
                               System.out.println("Msg:"+message.toString(Charset.defaultCharset()));

                               
                               ByteBuf response = ctx.alloc().buffer(20);
                               response.writeBytes("WriteBack__".getBytes(StandardCharsets.UTF_8));
                               response.writeBytes(message);
                               ctx.channel().writeAndFlush(response);
                           }
                        });
                    }
                }).bind(8080);

    }
}

EcoClient

package practice1.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;



public class EcoClient {

    public static void main(String[] args) throws InterruptedException {
        Channel channel = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline()
                                .addLast(new ChannelInboundHandlerAdapter() {
                                    @Override
                                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                        printf("From:" + ctx.channel().toString()+"n");
                                        ByteBuf buffer = msg instanceof ByteBuf ? (ByteBuf) msg : null;
                                        printf("Msg:" + buffer.toString(Charset.defaultCharset()) +"n");

                                    }
                                });

                    }
                }).connect(new InetSocketAddress("localhost", 8080))
                .sync()
                .channel();
        System.out.println("Client 启动成功");
        System.out.println("===========================");
        System.out.println("连接建立成功"+channel);
        new Thread(() ->{
            Scanner scanner = new Scanner(System.in);
            while (true){
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                printf("请输出>>>");
                String line = scanner.nextLine();
                if ("q".equals(line)){
                    System.out.println("程序运行结束！");
                    channel.close();
                    break;
                }
                byte[] bytes = line.getBytes(StandardCharsets.UTF_8);
                ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
                byteBuf.writeBytes(bytes);
                channel.writeAndFlush(byteBuf);
            }
        },"ClientInputThread").start();
    }

    private static synchronized void printf(String param){
        System.out.printf(param);
    }

}