《序列化与自定义Request、Response编解码器》

「爱情、让人受尽委屈。」 2022-02-23 02:50 499阅读 0赞

《序列化与自定义Request、Response编解码器》

  • 序列化是如何实现的
    • 序列化三种底层实现方式
      • 使用JDK的ByteArrayOutputStream序列化(需要自己手写大小端转字节序列的函数)
      • 用NIO中的Bytebuf(不能自动扩容)
      • 使用Netty中的ChannelBuffer
    • 序列化对象
      • 使用继承Serializer抽象类
      • 采用Protobuf序列化框架
  • 序列化协议在编解码器中的应用
    • 如何自定义写一个编解码器应用在数据传输中
      • 先实现数据包
      • Request的编解码器
        • RequestEncoder
        • RequestDecoder
      • response编解码器
        • ResponseEncoder
        • ResponseDecoder
      • 测试自定义编解码器

序列化是如何实现的

对象的序列化主要有两种用途：

1） 把对象的字节序列永久地保存到硬盘上，通常存放在一个文件中；

2） 在网络上传送对象的字节序列。

  在很多应用中，需要对某些对象进行序列化，让它们离开内存空间，入住物理硬盘，以便长期保存。比如最常见的是Web服务器中的Session对象，当有 10万用户并发访问，就有可能出现10万个Session对象，内存可能吃不消，于是Web容器就会把一些seesion先序列化到硬盘中，等要用了，再把保存在硬盘中的对象还原到内存中。
  当两个进程在进行远程通信时，彼此可以发送各种类型的数据。无论是何种类型的数据，都会以二进制序列的形式在网络上传送。发送方需要把这个Java对象转换为字节序列，才能在网络上传送；接收方则需要把字节序列再恢复为Java对象。
  java序列化是java编解码技术中的一种，不同的编解码的优缺点，参考之前的博客。https://blog.csdn.net/weixin_41262453/article/details/88980701

序列化三种底层实现方式

使用JDK的ByteArrayOutputStream序列化(需要自己手写大小端转字节序列的函数)

  ByteArrayOutputStream类实现了将数据写入字节数组的输出流。 当数据写入缓冲区时，缓冲区会自动增长。ByteArrayOutputStream.write(int b) 将指定的字节写入此字节数组输出流。 此方式需要自己手写大小端转字节序列的函数，麻烦。

  1. public class Test1 { 
  2.     public static void main(String[] args) throws IOException { 
  3.         int id = 101;
  4.         int age = 21;
  6.         ByteArrayOutputStream arrayOutputStream = new ByteArrayOutputStream();
  7.         arrayOutputStream.write(int2bytes(id));
  8.         arrayOutputStream.write(int2bytes(age));
  10.         byte[] byteArray = arrayOutputStream.toByteArray();
  11.         System.out.println(Arrays.toString(byteArray));
  13.         ByteArrayInputStream arrayInputStream = new ByteArrayInputStream(byteArray);
  14.         byte[] idBytes = new byte[4];
  15.         arrayInputStream.read(idBytes);
  16.         System.out.println("id:" + bytes2int(idBytes));
  18.         byte[] ageBytes = new byte[4];
  19.         arrayInputStream.read(ageBytes);
  20.         System.out.println("age:" + bytes2int(ageBytes));
  22.     }    
  23.     /** * 大端字节序列(先写高位，再写低位) * 百度下 大小端字节序列 * @param i * @return */
  24.     public static byte[] int2bytes(int i){ 
  25.         byte[] bytes = new byte[4];
  26.         bytes[0] = (byte)(i >> 3*8);
  27.         bytes[1] = (byte)(i >> 2*8);
  28.         bytes[2] = (byte)(i >> 1*8);
  29.         bytes[3] = (byte)(i >> 0*8);
  30.         return bytes;
  31.     }    
  32.     /** * 大端 * @param bytes * @return */
  33.     public static int bytes2int(byte[] bytes){ 
  34.         return (bytes[0] << 3*8) |
  35.                 (bytes[1] << 2*8) |
  36.                 (bytes[2] << 1*8) |
  37.                 (bytes[3] << 0*8);
  38.     }
  39. }

运行结果：
                在这里插入图片描述

用NIO中的Bytebuf(不能自动扩容)

  使用NIO中的字节缓冲区类Bytebuf，提供了许多序列化方法。使用方便，但是在使用时要定义大小，不能自动扩容。

  1. public class Test2 { 
  3.     public static void main(String[] args) { 
  4.         int id = 101;
  5.         int age = 21;
  7.         ByteBuffer buffer = ByteBuffer.allocate(8);
  8.         buffer.putInt(id);
  9.         buffer.putInt(age);
  10.         byte[] array = buffer.array();
  11.         System.out.println(Arrays.toString(buffer.array()));
  13.         //反序列化
  14.         ByteBuffer buffer2 = ByteBuffer.wrap(array);
  15.         System.out.println("id:"+buffer2.getInt());
  16.         System.out.println("age:"+buffer2.getInt());
  17.     }
  18. }

运行结果：
           在这里插入图片描述

使用Netty中的ChannelBuffer

  使用时需要导入Netty的jar包。

  1. public class Test3 { 
  2.     public static void main(String[] args) { 
  4.         ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();
  5.         buffer.writeInt(101);
  6.         buffer.writeDouble(80.1);
  8.         byte[] bytes = new byte[buffer.writerIndex()];
  9.         buffer.readBytes(bytes);
  11.         System.out.println(Arrays.toString(bytes));
  13.         "abc".getBytes();
  15.         //=====================反序列化===========================
  16.         ChannelBuffer wrappedBuffer = ChannelBuffers.wrappedBuffer(bytes);
  17.         System.out.println(wrappedBuffer.readInt());
  18.         System.out.println(wrappedBuffer.readDouble());        
  19.     }
  20. }

运行结果：
          在这里插入图片描述

序列化对象

使用继承Serializer抽象类

  Serializer底层的实现就是两个ChannelBuffer， writeBuffer和readBuffer。
  把对象转换为字节序列的过程称为对象的序列化；把字节序列恢复为对象的过程称为对象的反序列化。需要序列化的两个类：Player 和Resource ，继承Serializer抽象类要做的事情就是重写write和read方法。

  1. public class Player extends Serializer{ 
  3.     private long playerId;
  5.     private int age;
  7.     private List<Integer> skills = new ArrayList<>();
  9.     private Resource resource = new Resource();
  10.     public Resource getResource() { 
  11.         return resource;
  12.     }
  13.     public void setResource(Resource resource) { 
  14.         this.resource = resource;
  15.     }
  16.     public long getPlayerId() { 
  17.         return playerId;
  18.     }
  19.     public void setPlayerId(long playerId) { 
  20.         this.playerId = playerId;
  21.     }
  22.     public int getAge() { 
  23.         return age;
  24.     }
  25.     public void setAge(int age) { 
  26.         this.age = age;
  27.     }
  28.     public List<Integer> getSkills() { 
  29.         return skills;
  30.     }
  31.     public void setSkills(List<Integer> skills) { 
  32.         this.skills = skills;
  33.     }
  34.     @Override
  35.     protected void read() { 
  36.         this.playerId = readLong();
  37.         this.age = readInt();
  38.         this.skills = readList(Integer.class);
  39.         this.resource = read(Resource.class);
  40.     }
  41.     @Override
  42.     protected void write() { 
  43.         writeLong(playerId);
  44.         writeInt(age);
  45.         writeList(skills);
  46.         writeObject(resource);
  47.     }
  48. }
  49. public class Resource extends Serializer { 
  51.     private int gold;
  54.     public int getGold() { 
  55.         return gold;
  56.     }
  58.     public void setGold(int gold) { 
  59.         this.gold = gold;
  60.     }
  62.     @Override
  63.     protected void read() { 
  64.         this.gold = readInt();
  65.     }
  67.     @Override
  68.     protected void write() { 
  69.         writeInt(gold);
  70.     }
  71. }

测试序列化对象：

  1. public class Test4 { 
  3.     public static void main(String[] args) { 
  5.         Player player = new Player();
  6.         player.setPlayerId(10001);
  7.         player.setAge(22);
  8.         player.getSkills().add(101);
  9.         player.getResource().setGold(99999);
  11.         byte[] bytes = player.getBytes();
  13.         System.out.println(Arrays.toString(bytes));
  15.         //==============================================
  17.         Player player2 = new Player();
  18.         player2.readFromBytes(bytes);
  19.         System.out.println(player2.getPlayerId() + " "+player2.getAge() + " "+ Arrays.toString(player2.getSkills().toArray())+" " +player2.getResource().getGold());
  21.     }
  22. }

运行结果：
在这里插入图片描述

采用Protobuf序列化框架

  Protobuf系列化操作参考https://blog.csdn.net/weixin\_41262453/article/details/88980701\#Google\_Protobuf\_451 ，将上面的player序列化，进行测试。
  player.proto配置文件参考
              在这里插入图片描述
序列化后得到的PlayerModule有1000多行就不放了，测试代码如下：

  1. import java.util.Arrays;
  2. import com.proto.PlayerModule.PBPlayer;
  3. import com.proto.PlayerModule.PBPlayer.Builder;
  4. public class PB2Bytes { 
  5.     public static void main(String[] args) throws Exception { 
  6.         byte[] bytes = toBytes();
  7.         toPlayer(bytes);
  9.     }
  10.     /** * 序列化 */
  11.     public static byte[] toBytes(){ 
  12.         //获取一个PBPlayer的构造器
  13.         Builder builder = PlayerModule.PBPlayer.newBuilder();
  14.         //设置数据
  15.         builder.setPlayerId(101).setAge(20).setName("peter").addSkills(1001);
  16.         //构造出对象
  17.         PBPlayer player = builder.build();
  18.         //序列化成字节数组
  19.         byte[] byteArray = player.toByteArray();
  21.         System.out.println(Arrays.toString(byteArray));
  23.         return byteArray;
  24.     }    
  25.     /** * 反序列化 * @param bs * @throws Exception */
  26.     public static void toPlayer(byte[] bs) throws Exception{ 
  28.          PBPlayer player = PlayerModule.PBPlayer.parseFrom(bs);
  30.          System.out.println("playerId:" + player.getPlayerId());
  31.          System.out.println("age:" + player.getAge());
  32.          System.out.println("name:" + player.getName());
  33.          System.out.println("skills:" + (Arrays.toString(player.getSkillsList().toArray())));
  34.     }
  35. }

运行结果：
        在这里插入图片描述

序列化协议在编解码器中的应用

  此处提供了一个自定义的编解码器用于传输我们已经序列化的对象，可用于了解常见通信框架的编解码器底层序列化工作原理。

如何自定义写一个编解码器应用在数据传输中

先实现数据包

  首先设计数据包的格式：

  1. * 数据包格式
  2.  * +——----——+——-----——+——----——+——----——+——-----——+
  3.  * | 包头    | 模块号  | 命令号 |  长度  |   数据  |
  4.  * +——----——+——-----——+——----——+——----——+——-----——+
  5.  * </pre>
  6.  * 包头4字节
  7.  * 模块号2字节short
  8.  * 命令号2字节short
  9.  * 长度4字节(描述数据部分字节长度)

  因此我们要定义一个包头，首先采用一个不常用的4字节数据作为包头：

  1. public interface ConstantValue {     
  2.     /** * 包头 */
  3.     public static final int FLAG = -32523523;
  5. }

  其次定义请求，请求中包括请求模块、命令号、数据

  1. public class Request {     
  2.     /** * 请求模块 */
  3.     private short module;    
  4.     /** * 命令号 */
  5.     private short cmd;    
  6.     /** * 数据部分 */
  7.     private byte[] data;
  8.     public short getModule() { 
  9.         return module;
  10.     }
  11.     public void setModule(short module) { 
  12.         this.module = module;
  13.     }
  14.     public short getCmd() { 
  15.         return cmd;
  16.     }
  17.     public void setCmd(short cmd) { 
  18.         this.cmd = cmd;
  19.     }
  20.     public byte[] getData() { 
  21.         return data;
  22.     }
  23.     public void setData(byte[] data) { 
  24.         this.data = data;
  25.     }
  26.     public int getDataLength(){ 
  27.         if(data == null){ 
  28.             return 0;
  29.         }
  30.         return data.length;
  31.     }
  32. }

响应的有响应：Response

  1. public class Response { 
  2.     /** * 请求模块 */
  3.     private short module;    
  4.     /** * 命令号 */
  5.     private short cmd;    
  6.     /** * 状态码 */
  7.     private int stateCode;    
  8.     /** * 数据部分 */
  9.     private byte[] data;
  10.     public short getModule() { 
  11.         return module;
  12.     }
  13.     public void setModule(short module) { 
  14.         this.module = module;
  15.     }
  16.     public short getCmd() { 
  17.         return cmd;
  18.     }
  19.     public void setCmd(short cmd) { 
  20.         this.cmd = cmd;
  21.     }
  22.     public int getStateCode() { 
  23.         return stateCode;
  24.     }
  25.     public void setStateCode(int stateCode) { 
  26.         this.stateCode = stateCode;
  27.     }
  28.     public byte[] getData() { 
  29.         return data;
  30.     }
  31.     public void setData(byte[] data) { 
  32.         this.data = data;
  33.     }
  34.     public int getDataLength(){ 
  35.         if(data == null){ 
  36.             return 0;
  37.         }
  38.         return data.length;
  39.     }
  40. }

返回的状态字：

  1. public interface StateCode {     
  2.     /** * 成功 */
  3.     public static int SUCCESS  = 0;
  5.     /** * 失败 */
  6.     public static int FAIL  =  1;
  7. }

Request的编解码器

RequestEncoder

  RequestEncoder首先需要继承Netty3.10.5中OneToOneEncoder，RequestEncoder将请求Request对象实现序列化到ChannelBuffer ，返回ChannelBuffer 缓存区buffer。

  1. /** * 请求编码器 * <pre> * 数据包格式 * +——----——+——-----——+——----——+——----——+——-----——+ * | 包头 | 模块号 | 命令号 | 长度 | 数据 | * +——----——+——-----——+——----——+——----——+——-----——+ * </pre> * 包头4字节 * 模块号2字节short * 命令号2字节short * 长度4字节(描述数据部分字节长度) */
  2. public class RequestEncoder extends OneToOneEncoder{ 
  3.     @Override
  4.     protected Object encode(ChannelHandlerContext context, Channel channel, Object rs) throws Exception { 
  5.         Request request = (Request)(rs);        
  6.         ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();
  7.         //包头
  8.         buffer.writeInt(ConstantValue.FLAG);
  9.         //module
  10.         buffer.writeShort(request.getModule());
  11.         //cmd
  12.         buffer.writeShort(request.getCmd());
  13.         //长度
  14.         buffer.writeInt(request.getDataLength());
  15.         //data
  16.         if(request.getData() != null){ 
  17.             buffer.writeBytes(request.getData());
  18.         }        
  19.         return buffer;
  20.     }
  21. }

RequestDecoder

  RequestDecoder同样要继承Netty3.10.5中FrameDecoder，Request解码器要做的事情就是从接收到的字节流缓存区读取出数据，进行判断，若包头正确，可读长度大于基本长度，则将字节流依次读取出Request类的各个成员变量，返回Request对象。

  1. /** * 请求解码器 * <pre> * 数据包格式 * +——----——+——-----——+——----——+——----——+——-----——+ * | 包头 | 模块号 | 命令号 | 长度 | 数据 | * +——----——+——-----——+——----——+——----——+——-----——+ * </pre> * 包头4字节 * 模块号2字节short * 命令号2字节short * 长度4字节(描述数据部分字节长度) * * */
  2. public class RequestDecoder extends FrameDecoder{     
  3.     /** * 数据包基本长度 */
  4.     public static int BASE_LENTH = 4 + 2 + 2 + 4;
  5.     @Override
  6.     protected Object decode(ChannelHandlerContext arg0, Channel arg1, ChannelBuffer buffer) throws Exception {         
  7.         //可读长度必须大于基本长度
  8.         if(buffer.readableBytes() >= BASE_LENTH){ 
  9.             //防止socket字节流攻击
  10.             if(buffer.readableBytes() > 2048){ 
  11.                 buffer.skipBytes(buffer.readableBytes());
  12.             }            
  13.             //记录包头开始的index
  14.             int beginReader;
  16.             while(true){ 
  17.                 beginReader = buffer.readerIndex();
  18.                 buffer.markReaderIndex();
  19.                 if(buffer.readInt() == ConstantValue.FLAG){ 
  20.                     break;
  21.                 }                
  22.                 //未读到包头，略过一个字节
  23.                 buffer.resetReaderIndex();
  24.                 buffer.readByte();
  26.                 //长度又变得不满足
  27.                 if(buffer.readableBytes() < BASE_LENTH){ 
  28.                     return null;
  29.                 }
  30.             }            
  31.             //模块号
  32.             short module = buffer.readShort();
  33.             //命令号
  34.             short cmd = buffer.readShort();
  35.             //长度
  36.             int length = buffer.readInt();
  38.             //判断请求数据包数据是否到齐
  39.             if(buffer.readableBytes() < length){ 
  40.                 //还原读指针
  41.                 buffer.readerIndex(beginReader);
  42.                 return null;
  43.             }            
  44.             //读取data数据
  45.             byte[] data = new byte[length];
  46.             buffer.readBytes(data);            
  47.             Request request = new Request();
  48.             request.setModule(module);
  49.             request.setCmd(cmd);
  50.             request.setData(data);            
  51.             //继续往下传递 
  52.             return request;            
  53.         }
  54.         //数据包不完整，需要等待后面的包来
  55.         return null;
  56.     }
  57. }

response编解码器

ResponseEncoder

  ResponseEncoder与RequestEncoder实现原理类似。

  1. /** * 请求编码器 * <pre> * 数据包格式 * +——----——+——-----——+——----——+——----——+——-----——+——-----——+ * | 包头 | 模块号 | 命令号 | 状态码 | 长度 | 数据 | * +——----——+——-----——+——----——+——----——+——-----——+——-----——+ * </pre> * 包头4字节 * 模块号2字节short * 命令号2字节short * 长度4字节(描述数据部分字节长度) */
  2. public class ResponseEncoder extends OneToOneEncoder{ 
  3.     @Override
  4.     protected Object encode(ChannelHandlerContext context, Channel channel, Object rs) throws Exception { 
  5.         Response response = (Response)(rs);
  7.         ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();
  8.         //包头
  9.         buffer.writeInt(ConstantValue.FLAG);
  10.         //module
  11.         buffer.writeShort(response.getModule());
  12.         //cmd
  13.         buffer.writeShort(response.getCmd());
  14.         //状态码
  15.         buffer.writeInt(response.getStateCode());
  16.         //长度
  17.         buffer.writeInt(response.getDataLength());
  18.         //data
  19.         if(response.getData() != null){ 
  20.             buffer.writeBytes(response.getData());
  21.         }    
  22.         return buffer;
  23.     }
  24. }

ResponseDecoder

  ResponseDecoder相比RequestDecoder，还会多解析一个状态码。

  1. /** * response解码器 * <pre> * 数据包格式 * +——----——+——-----——+——----——+——----——+——-----——+——-----——+ * | 包头 | 模块号 | 命令号 | 状态码 | 长度 | 数据 | * +——----——+——-----——+——----——+——----——+——-----——+——-----——+ * </pre> * 包头4字节 * 模块号2字节short * 命令号2字节short * 长度4字节(描述数据部分字节长度) */
  2. public class ResponseDecoder extends FrameDecoder{     
  3.     /** * 数据包基本长度 */
  4.     public static int BASE_LENTH = 4 + 2 + 2 + 4;
  5.     @Override
  6.     protected Object decode(ChannelHandlerContext arg0, Channel arg1, ChannelBuffer buffer) throws Exception {         
  7.         //可读长度必须大于基本长度
  8.         if(buffer.readableBytes() >= BASE_LENTH){ 
  10.             //记录包头开始的index
  11.             int beginReader = buffer.readerIndex();
  13.             while(true){ 
  14.                 if(buffer.readInt() == ConstantValue.FLAG){ 
  15.                     break;
  16.                 }
  17.             }
  18.             //模块号
  19.             short module = buffer.readShort();
  20.             //命令号
  21.             short cmd = buffer.readShort();
  22.             //状态码
  23.             int stateCode = buffer.readInt();
  24.             //长度
  25.             int length = buffer.readInt();
  27.             if(buffer.readableBytes() < length){ 
  28.                 //还原读指针
  29.                 buffer.readerIndex(beginReader);
  30.                 return null;
  31.             }            
  32.             byte[] data = new byte[length];
  33.             buffer.readBytes(data);            
  34.             Response response = new Response();
  35.             response.setModule(module);
  36.             response.setCmd(cmd);
  37.             response.setStateCode(stateCode);
  38.             response.setData(data);            
  39.             //继续往下传递 
  40.             return response;            
  41.         }
  42.         //数据包不完整，需要等待后面的包来
  43.         return null;
  44.     }
  45. }

测试自定义编解码器

  发送的对象模型：FightRequest

  1. public class FightRequest extends Serializer{     
  2.     /** * 副本id */
  3.     private int fubenId;    
  4.     /** * 次数 */
  5.     private int count;
  7.     public int getFubenId() { 
  8.         return fubenId;
  9.     }
  10.     public void setFubenId(int fubenId) { 
  11.         this.fubenId = fubenId;
  12.     }
  13.     public int getCount() { 
  14.         return count;
  15.     }
  16.     public void setCount(int count) { 
  17.         this.count = count;
  18.     }
  19.     @Override
  20.     protected void read() { 
  21.         this.fubenId = readInt();
  22.         this.count = readInt();
  23.     }
  24.     @Override
  25.     protected void write() { 
  26.         writeInt(fubenId);
  27.         writeInt(count);
  28.     }
  29. }

  返回的对象模型：

  1. public class FightResponse extends Serializer{ 
  2.     /** * 获取金币 */
  3.     private int gold;
  4.     public int getGold() { 
  5.         return gold;
  6.     }
  7.     public void setGold(int gold) { 
  8.         this.gold = gold;
  9.     }
  10.     @Override
  11.     protected void read() { 
  12.         this.gold = readInt();
  13.     }
  14.     @Override
  15.     protected void write() { 
  16.         writeInt(gold);
  17.     }
  18. }

  测试代码：服务端Client和HiHandler ,由于Request和Response的Encoder和Decoder都是继承了Netty中的类进行设计的，所以可以直接pipeline.addLast，因此兼容原有的Netty框架。并且在channel.write(request)时也是直接write request对象，request对象中的data就是需要发送的fightRequest模型的序列化字节码。

  1. public class Client { 
  2.     public static void main(String[] args) throws InterruptedException {         
  3.         //服务类
  4.         ClientBootstrap bootstrap = new  ClientBootstrap();
  6.         //线程池
  7.         ExecutorService boss = Executors.newCachedThreadPool();
  8.         ExecutorService worker = Executors.newCachedThreadPool();        
  9.         //socket工厂
  10.         bootstrap.setFactory(new NioClientSocketChannelFactory(boss, worker));
  12.         //管道工厂
  13.         bootstrap.setPipelineFactory(new ChannelPipelineFactory() {             
  14.             @Override
  15.             public ChannelPipeline getPipeline() throws Exception { 
  16.                 ChannelPipeline pipeline = Channels.pipeline();
  17.                 pipeline.addLast("decoder", new ResponseDecoder());
  18.                 pipeline.addLast("encoder", new RequestEncoder());
  19.                 pipeline.addLast("hiHandler", new HiHandler());
  20.                 return pipeline;
  21.             }
  22.         });        
  23.         //连接服务端
  24.         ChannelFuture connect = bootstrap.connect(new InetSocketAddress("127.0.0.1", 30000));
  25.         Channel channel = connect.sync().getChannel();        
  26.         System.out.println("client start");        
  27.         Scanner scanner = new Scanner(System.in);
  28.         while(true){ 
  29.             System.out.println("请输入");
  30.             int fubenId = Integer.parseInt(scanner.nextLine());
  31.             int count = Integer.parseInt(scanner.nextLine());
  33.             FightRequest fightRequest = new FightRequest();
  34.             fightRequest.setFubenId(fubenId);
  35.             fightRequest.setCount(count);
  37.             Request request = new Request();
  38.             request.setModule((short) 1);
  39.             request.setCmd((short) 1);
  40.             request.setData(fightRequest.getBytes());
  41.             //发送请求
  42.             channel.write(request);
  43.         }
  44.     }
  45. }
  47. public class HiHandler extends SimpleChannelHandler { 
  48.     /** * 接收消息 */
  49.     @Override
  50.     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { 
  51.             Response message = (Response)e.getMessage();
  53.             if(message.getModule() == 1){ 
  55.                 if(message.getCmd() == 1){ 
  56.                     FightResponse fightResponse = new FightResponse();
  57.                     fightResponse.readFromBytes(message.getData());
  59.                     System.out.println("gold:" + fightResponse.getGold());
  61.                 }else if(message.getCmd() == 2){ 
  63.                 }
  65.             }else if (message.getModule() == 1){ 
  68.             }
  69.     }
  70.     /** * 捕获异常 */
  71.     @Override
  72.     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { 
  73.         System.out.println("exceptionCaught");
  74.         super.exceptionCaught(ctx, e);
  75.     }
  76.     /** * 新连接 */
  77.     @Override
  78.     public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { 
  79.         System.out.println("channelConnected");
  80.         super.channelConnected(ctx, e);
  81.     }
  82.     /** * 必须是链接已经建立，关闭通道的时候才会触发 */
  83.     @Override
  84.     public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { 
  85.         System.out.println("channelDisconnected");
  86.         super.channelDisconnected(ctx, e);
  87.     }
  88.     /** * channel关闭的时候触发 */
  89.     @Override
  90.     public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { 
  91.         System.out.println("channelClosed");
  92.         super.channelClosed(ctx, e);
  93.     }
  94. }

  客户端Server 和HelloHandler：messageReceived中接收的消息此时就是Request对象，从Request对象的data取出的数据是序列化的fightRequest 字节码，要再经过该类中的readFromBytes方法转换，才能完全转成发送过来的fightRequest 。

  1. public class Server { 
  2.     public static void main(String[] args) { 
  3.         //服务类
  4.         ServerBootstrap bootstrap = new ServerBootstrap();        
  5.         //boss线程监听端口，worker线程负责数据读写
  6.         ExecutorService boss = Executors.newCachedThreadPool();
  7.         ExecutorService worker = Executors.newCachedThreadPool();        
  8.         //设置niosocket工厂
  9.         bootstrap.setFactory(new NioServerSocketChannelFactory(boss, worker));        
  10.         //设置管道的工厂
  11.         bootstrap.setPipelineFactory(new ChannelPipelineFactory() { 
  13.             @Override
  14.             public ChannelPipeline getPipeline() throws Exception { 
  16.                 ChannelPipeline pipeline = Channels.pipeline();
  17.                 pipeline.addLast("decoder", new RequestDecoder());
  18.                 pipeline.addLast("encoder", new ResponseEncoder());
  19.                 pipeline.addLast("helloHandler", new HelloHandler());
  20.                 return pipeline;
  21.             }
  22.         });        
  23.         bootstrap.bind(new InetSocketAddress(30000));        
  24.         System.out.println("start!!!");        
  25.     }
  26. }
  28. public class HelloHandler extends SimpleChannelHandler { 
  30.     /** * 接收消息 */
  31.     @Override
  32.     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { 
  33.         Request message = (Request)e.getMessage();        
  34.         if(message.getModule() == 1){ 
  35.                 if(message.getCmd() == 1){             
  36.                 FightRequest fightRequest = new FightRequest();
  37.                 fightRequest.readFromBytes(message.getData());
  39.                 System.out.println("fubenId:" +fightRequest.getFubenId() + " " + "count:" + fightRequest.getCount());
  41.                 //回写数据
  42.                 FightResponse fightResponse = new FightResponse();
  43.                 fightResponse.setGold(9999);
  45.                 Response response = new Response();
  46.                 response.setModule((short) 1);
  47.                 response.setCmd((short) 1);
  48.                 response.setStateCode(StateCode.SUCCESS);
  49.                 response.setData(fightResponse.getBytes());
  50.                 ctx.getChannel().write(response);
  51.             }else if(message.getCmd() == 2){                 
  52.             }            
  53.         }else if (message.getModule() == 1){ 
  55.         }
  56.     }
  57.     /** * 捕获异常 */
  58.     @Override
  59.     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { 
  60.         System.out.println("exceptionCaught");
  61.         super.exceptionCaught(ctx, e);
  62.     }
  63.     /** * 新连接 */
  64.     @Override
  65.     public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { 
  66.         System.out.println("channelConnected");
  67.         super.channelConnected(ctx, e);
  68.     }
  69.     /** * 必须是链接已经建立，关闭通道的时候才会触发 */
  70.     @Override
  71.     public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { 
  72.         System.out.println("channelDisconnected");
  73.         super.channelDisconnected(ctx, e);
  74.     }
  75.     /** * channel关闭的时候触发 */
  76.     @Override
  77.     public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { 
  78.         System.out.println("channelClosed");
  79.         super.channelClosed(ctx, e);
  80.     }
  81. }

运行测试：
              在这里插入图片描述
服务器端：
              在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，499人围观）

还没有评论，来说两句吧...

相关阅读