NETTY序列化与反序列化

@Sharable
public class ObjectEncoder extends MessageToByteEncoder<Serializable> {
    private static final byte[] LENGTH_PLACEHOLDER = new byte[4];
    //当需要编码时，encode方法会被回调 
    //参数msg：就是我们需要序列化的java对象
    //参数out：我们需要将序列化后的二进制字节写到ByteBuf中
    @Override
    protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out) throws Exception {
        int startIdx = out.writerIndex();
        //ByteBufOutputStream是Netty提供的输出流，数据写入其中之后，可以通过其buffer()方法会的对应的ByteBuf实例
        ByteBufOutputStream bout = new ByteBufOutputStream(out);
        //JDK序列化机制的ObjectOutputStream
        ObjectOutputStream oout = null;
        try {
            //首先占用4个字节，这就是Length字段的字节数，这只是占位符，后面为填充对象序列化后的字节数
            bout.write(LENGTH_PLACEHOLDER);
            //CompactObjectOutputStream是netty提供的类，其实现了JDK的ObjectOutputStream，顾名思义用于压缩
            //同时把bout作为其底层输出流，意味着对象序列化后的字节直接写到了bout中
            oout = new CompactObjectOutputStream(bout);
            //调用writeObject方法，即表示开始序列化
            oout.writeObject(msg);
            oout.flush();
        } finally {
            if (oout != null) {
                oout.close();
            } else {
                bout.close();
            }
        }
        int endIdx = out.writerIndex();
        //序列化完成，设置占位符的值，也就是对象序列化后的字节数量
        out.setInt(startIdx, endIdx - startIdx - 4);
    }
}

//注意ObjectDecoder继承了LengthFieldBasedFrameDecoder
public class ObjectDecoder extends LengthFieldBasedFrameDecoder {
    
    private final ClassResolver classResolver;
    public ObjectDecoder(ClassResolver classResolver) {
        this(1048576, classResolver);
    }
    //参数maxObjectSize：表示可接受的对象反序列化的最大字节数，默认为1048576 bytes，约等于1M
    //参数classResolver：由于需要将二进制字节反序列化为Java对象，需要指定一个ClassResolver来加载这个类的字节码对象
    public ObjectDecoder(int maxObjectSize, ClassResolver classResolver) {
        //调用父类LengthFieldBasedFrameDecoder构造方法，关于这几个参数的作用，参见之前章节的分析
        super(maxObjectSize, 0, 4, 0, 4);
        this.classResolver = classResolver;
    }
    //当需要解码时，decode方法会被回调
    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        //首先调用父类的decode方法进行解码，会解析出包含可解析为java对象的完整二进制字节封装到ByteBuf中，同时Length字段的4个字节会被删除
        ByteBuf frame = (ByteBuf) super.decode(ctx, in);
        if (frame == null) {
            return null;
        }
        //构造JDK ObjectInputStream实例用于解码
        ObjectInputStream ois = new CompactObjectInputStream(new ByteBufInputStream(frame, true), classResolver);
        try {
            //调用readObject方法进行解码，其返回的就是反序列化之后的Java对象
            return ois.readObject();
        } finally {
            ois.close();
        }
    }
}

public class Request implements Serializable{
    private String request;
    private Date requestTime;
    //setters getters and toString
}
public class Response implements Serializable{
    private String response;
    private Date responseTime;
    //setters getters and toString
}

public class JdkSerializerClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap(); // (1)
            b.group(workerGroup); // (2)
            b.channel(NioSocketChannel.class); // (3)
            b.option(ChannelOption.SO_KEEPALIVE, true); // (4)
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new ObjectEncoder());
                    ch.pipeline().addLast(new ObjectDecoder(new ClassResolver() {
                        public Class<?> resolve(String className) throws ClassNotFoundException {
                            return Class.forName(className);
                        }
                    }));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        // 在于server建立连接后，即发送请求报文
                        public void channelActive(ChannelHandlerContext ctx) {
                            Request request = new Request();
                            request.setRequest("i am request!");
                            request.setRequestTime(new Date());
                            ctx.writeAndFlush(request);
                        }
                        //接受服务端的响应
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            Response response= (Response) msg;
                            System.out.println("receive response:"+response);
                        }
                    });
                }
            });
            // Start the client.
            ChannelFuture f = b.connect("127.0.0.1", 8080).sync(); // (5)
            // Wait until the connection is closed.
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}

public class JdkSerializerServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap(); // (2)
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) // (3)
                    .childHandler(new ChannelInitializer<SocketChannel>() { // (4)
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new ObjectEncoder());
                            ch.pipeline().addLast(new ObjectDecoder(new ClassResolver() {
                                public Class<?> resolve(String className) throws ClassNotFoundException {
                                    return Class.forName(className);
                                }
                            }));
                            // 自定义这个ChannelInboundHandler打印拆包后的结果
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                    Request request= (Request) msg;
                                    System.out.println("receive request:"+request);
                                    Response response = new Response();
                                    response.setResponse("response to:"+request.getRequest());
                                    response.setResponseTime(new Date());
                                    ctx.writeAndFlush(response);
                                }
                            });
                        }
                    });
            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(8080).sync(); // (7)
            System.out.println("JdkSerializerServer Started on 8080...");
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

package tutorial;
 
option java_package = "com.example.tutorial";
option java_outer_classname = "AddressBookProtos";
 
message AddressBook {
    repeated Person person = 1;
}
 
message Person {
    required string name = 1;
    required int32 id = 2;
    optional string email = 3;
 
    enum PhoneType {
        MOBILE = 0;
        HOME = 1;
        WORK = 2;
    }
 
    message PhoneNumber {
        required string number = 1;
        optional PhoneType type = 2 [default = HOME];
    }
 
    repeated PhoneNumber phone = 4;
}

// required string name = 1;
public boolean hasName();
public String getName();
 
// required int32 id = 2;
public boolean hasId();
public int getId();
 
// optional string email = 3;
public boolean hasEmail();
public String getEmail();
 
// repeated .tutorial.Person.PhoneNumber phone = 4;
public List<PhoneNumber> getPhoneList();
public int getPhoneCount();
public PhoneNumber getPhone(int index);

// required string name = 1;
public boolean hasName();
public java.lang.String getName();
public Builder setName(String value);
public Builder clearName();
 
// required int32 id = 2;
public boolean hasId();
public int getId();
public Builder setId(int value);
public Builder clearId();
 
// optional string email = 3;
public boolean hasEmail();
public String getEmail();
public Builder setEmail(String value);
public Builder clearEmail();
 
// repeated .tutorial.Person.PhoneNumber phone = 4;
public List<PhoneNumber> getPhoneList();
public int getPhoneCount();
public PhoneNumber getPhone(int index);
public Builder setPhone(int index, PhoneNumber value);
public Builder addPhone(PhoneNumber value);
public Builder addAllPhone(Iterable<PhoneNumber> value);
public Builder clearPhone();

public static enum PhoneType {
  MOBILE(0, 0),
  HOME(1, 1),
  WORK(2, 2),
  ;
  ...
}

Person john =
  Person.newBuilder()
    .setId(1234)
    .setName("John Doe")
    .setEmail("jdoe@example.com")
    .addPhone(
      Person.PhoneNumber.newBuilder()
        .setNumber("555-4321")
        .setType(Person.PhoneType.HOME))
    .build();

import com.example.tutorial.AddressBookProtos.AddressBook;
import com.example.tutorial.AddressBookProtos.Person;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.InputStreamReader;
import java.io.IOException;
import java.io.PrintStream;
 
class AddPerson {
  // This function fills in a Person message based on user input.
  static Person PromptForAddress(BufferedReader stdin,
                                 PrintStream stdout) throws IOException {
    Person.Builder person = Person.newBuilder();
 
    stdout.print("Enter person ID: ");
    person.setId(Integer.valueOf(stdin.readLine()));
 
    stdout.print("Enter name: ");
    person.setName(stdin.readLine());
 
    stdout.print("Enter email address (blank for none): ");
    String email = stdin.readLine();
    if (email.length() > 0) {
      person.setEmail(email);
    }
 
    while (true) {
      stdout.print("Enter a phone number (or leave blank to finish): ");
      String number = stdin.readLine();
      if (number.length() == 0) {
        break;
      }
 
      Person.PhoneNumber.Builder phoneNumber =
        Person.PhoneNumber.newBuilder().setNumber(number);
 
      stdout.print("Is this a mobile, home, or work phone? ");
      String type = stdin.readLine();
      if (type.equals("mobile")) {
        phoneNumber.setType(Person.PhoneType.MOBILE);
      } else if (type.equals("home")) {
        phoneNumber.setType(Person.PhoneType.HOME);
      } else if (type.equals("work")) {
        phoneNumber.setType(Person.PhoneType.WORK);
      } else {
        stdout.println("Unknown phone type.  Using default.");
      }
 
      person.addPhone(phoneNumber);
    }
 
    return person.build();
  }
 
  // Main function:  Reads the entire address book from a file,
  //   adds one person based on user input, then writes it back out to the same
  //   file.
  public static void main(String[] args) throws Exception {
    if (args.length != 1) {
      System.err.println("Usage:  AddPerson ADDRESS_BOOK_FILE");
      System.exit(-1);
    }
 
    AddressBook.Builder addressBook = AddressBook.newBuilder();
 
    // Read the existing address book.
    try {
      addressBook.mergeFrom(new FileInputStream(args[0]));
    } catch (FileNotFoundException e) {
      System.out.println(args[0] + ": File not found.  Creating a new file.");
    }
 
    // Add an address.
    addressBook.addPerson(
      PromptForAddress(new BufferedReader(new InputStreamReader(System.in)),
                       System.out));
 
    // Write the new address book back to disk.
    FileOutputStream output = new FileOutputStream(args[0]);
    addressBook.build().writeTo(output);
    output.close();
  }
}

import com.example.tutorial.AddressBookProtos.AddressBook;
import com.example.tutorial.AddressBookProtos.Person;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.PrintStream;
 
class ListPeople {
  // Iterates though all people in the AddressBook and prints info about them.
  static void Print(AddressBook addressBook) {
    for (Person person: addressBook.getPersonList()) {
      System.out.println("Person ID: " + person.getId());
      System.out.println("  Name: " + person.getName());
      if (person.hasEmail()) {
        System.out.println("  E-mail address: " + person.getEmail());
      }
 
      for (Person.PhoneNumber phoneNumber : person.getPhoneList()) {
        switch (phoneNumber.getType()) {
          case MOBILE:
            System.out.print("  Mobile phone #: ");
            break;
          case HOME:
            System.out.print("  Home phone #: ");
            break;
          case WORK:
            System.out.print("  Work phone #: ");
            break;
        }
        System.out.println(phoneNumber.getNumber());
      }
    }
  }
 
  // Main function:  Reads the entire address book from a file and prints all
  //   the information inside.
  public static void main(String[] args) throws Exception {
    if (args.length != 1) {
      System.err.println("Usage:  ListPeople ADDRESS_BOOK_FILE");
      System.exit(-1);
    }
 
    // Read the existing address book.
    AddressBook addressBook =
      AddressBook.parseFrom(new FileInputStream(args[0]));
 
    Print(addressBook);
  }
}

b.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .option(ChannelOption.SO_BACKLOG, 100)
        .handler(new LoggingHandler(LogLevel.INFO))
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) {
                ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                ch.pipeline().addLast(new ProtobufDecoder(AddressBookProtos.Person.getDefaultInstance()));
                ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                ch.pipeline().addLast(new ProtobufEncoder());
                ch.pipeline().addLast(new AddressBookHandler());
            }
        });

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
        throws Exception {
    
    AddressBookProtos.Person req = (AddressBookProtos.Person) msg;
    //...处理req
    //返回响应
    ctx.writeAndFlush(AddressBookProtos.AddressBook.newBuilder().
                            addPerson(AddressBookProtos.Person.newBuilder()
                            .setId(req.getId())
                            .setName(req.getName())
                            .setEmail("tianshouzhi@126.com")
                            .build()));
    }
}