netty8---自定义编码解码器

package com.cn.codc;

import org.jboss.netty.buffer.ChannelBuffer;

import org.jboss.netty.channel.Channel;

import org.jboss.netty.channel.ChannelHandlerContext;

import org.jboss.netty.handler.codec.frame.FrameDecoder;

import com.cn.constant.ConstantValue;

import com.cn.model.Request;

/**

 * 请求解码器

 * <pre>

 * 数据包格式

 * +——----——+——-----——+——----——+——----——+——-----——+

 * | 包头          | 模块号        | 命令号      |  长度        |   数据       |

 * +——----——+——-----——+——----——+——----——+——-----——+

 * </pre>

 * 包头4字节

 * 模块号2字节short

 * 命令号2字节short

 * 长度4字节(描述数据部分字节长度)

 */

public class RequestDecoder extends FrameDecoder{// FrameDecoder 这个decoder可以协助我们解决粘包分包问题

    /**

     * 数据包基本长度

     */

    public static int BASE_LENTH = 4 + 2 + 2 + 4;

    //ChannelBuffer里面有一个读指针和写指针。读指针和写指针初始值是0，写多少数据写指针就移动多少

    //调用readShort方法，readInt方法就会移动读指针， 0 =< readerIndex =< writerIndex

    @Override

    protected Object decode(ChannelHandlerContext arg0, Channel arg1, ChannelBuffer buffer) throws Exception {

        //可读长度readableBytes必须大于基本长度才处理

        if(buffer.readableBytes() >= BASE_LENTH){

            //防止socket字节流攻击

            if(buffer.readableBytes() > 2048){

                buffer.skipBytes(buffer.readableBytes());

            }

            //记录包头开始的index

            int beginReader;

            while(true){//循环读取，直到包头读取完毕

                beginReader = buffer.readerIndex();//获取读指针

                buffer.markReaderIndex();

                if(buffer.readInt() == ConstantValue.FLAG){

                    break;

                }

                //未读到包头，略过一个字节

                buffer.resetReaderIndex();

                buffer.readByte();

                //长度又变得不满足

                if(buffer.readableBytes() < BASE_LENTH){

                    return null;

                }

            }

            //包头读取完毕，读取模块号

            short module = buffer.readShort();

            //读取命令号

            short cmd = buffer.readShort();

            //读取长度

            int length = buffer.readInt();

            //readableBytes现在可读的长度小于数据的长度。判断请求数据包数据部分是否到齐

            if(buffer.readableBytes() < length){

                //还原读指针，已经读取了12个字节，但是没用，所以要还原buffer的读指针，

                buffer.readerIndex(beginReader);

                return null;//等待后面的数据包来

            }

            //比length要长，就读取data数据

            byte[] data = new byte[length];

            buffer.readBytes(data);//数据读取完毕

            //封装request对象继续向下传递

            Request request = new Request();

            request.setModule(module);

            request.setCmd(cmd);

            request.setData(data);

            //继续往下传递 ，调用sendUpStreamEvent方法向下传递

            return request;

        }

        //长度短了，数据包不完整，需要等待后面的包来

        return null;

        //FrameDecoder： return null就是等待后面的包，return一个解码的对象就是向下传递。

    }

}

package com.cn.codc;

import org.jboss.netty.buffer.ChannelBuffer;

import org.jboss.netty.buffer.ChannelBuffers;

import org.jboss.netty.channel.Channel;

import org.jboss.netty.channel.ChannelHandlerContext;

import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;

import com.cn.constant.ConstantValue;

import com.cn.model.Request;

/**

 * 请求编码器

 * <pre>

 * 数据包格式

 * +——----——+——-----——+——----——+——----——+——-----——+

 * | 包头          | 模块号        | 命令号      |  长度        |   数据       |

 * +——----——+——-----——+——----——+——----——+——-----——+

 * </pre>

 * 包头4字节

 * 模块号2字节short

 * 命令号2字节short

 * 长度4字节(描述数据部分字节长度)

 */

public class RequestEncoder extends OneToOneEncoder{

    //把一个request对象转换成了一个ChannelBuffer二进制数据

    @Override

    protected Object encode(ChannelHandlerContext context, Channel channel, Object rs) throws Exception {

        Request request = (Request)(rs);

        ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();

        //包头，确定数据包的开始

        buffer.writeInt(ConstantValue.FLAG);

        //module

        buffer.writeShort(request.getModule());

        //cmd

        buffer.writeShort(request.getCmd());

        //长度

        buffer.writeInt(request.getDataLength());

        //data

        if(request.getData() != null){

            buffer.writeBytes(request.getData());

        }

        return buffer;//返回一个ChannelBuffer继续向下传递。

    }

}

package com.cn.codc;

import org.jboss.netty.buffer.ChannelBuffer;

import org.jboss.netty.channel.Channel;

import org.jboss.netty.channel.ChannelHandlerContext;

import org.jboss.netty.handler.codec.frame.FrameDecoder;

import com.cn.constant.ConstantValue;

import com.cn.model.Response;

/**

 * response解码器

 * <pre>

 * 数据包格式

 * +——----——+——-----——+——----——+——----——+——-----——+——-----——+

 * | 包头          | 模块号        | 命令号       |  状态码    |  长度          |   数据       |

 * +——----——+——-----——+——----——+——----——+——-----——+——-----——+

 * </pre>

 * 包头4字节

 * 模块号2字节short

 * 命令号2字节short

 * 长度4字节(描述数据部分字节长度)

 */

public class ResponseDecoder extends FrameDecoder{

    /**

     * 数据包基本长度

     */

    public static int BASE_LENTH = 4 + 2 + 2 + 4;

    @Override

    protected Object decode(ChannelHandlerContext arg0, Channel arg1, ChannelBuffer buffer) throws Exception {

        //可读长度必须大于基本长度

        if(buffer.readableBytes() >= BASE_LENTH){

            //记录包头开始的index

            int beginReader = buffer.readerIndex();

            while(true){

                if(buffer.readInt() == ConstantValue.FLAG){

                    break;

                }

            }

            //模块号

            short module = buffer.readShort();

            //命令号

            short cmd = buffer.readShort();

            //状态码

            int stateCode = buffer.readInt();

            //长度

            int length = buffer.readInt();

            if(buffer.readableBytes() < length){

                //还原读指针

                buffer.readerIndex(beginReader);

                return null;

            }

            byte[] data = new byte[length];

            buffer.readBytes(data);

            //封装Response对象

            Response response = new Response();

            response.setModule(module);

            response.setCmd(cmd);

            response.setStateCode(stateCode);

            response.setData(data);

            //继续往下传递

            return response;

        }

        //数据包不完整，需要等待后面的包来

        return null;

    }

}

package com.cn.codc;

import org.jboss.netty.buffer.ChannelBuffer;

import org.jboss.netty.buffer.ChannelBuffers;

import org.jboss.netty.channel.Channel;

import org.jboss.netty.channel.ChannelHandlerContext;

import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;

import com.cn.constant.ConstantValue;

import com.cn.model.Response;

/**

 * 请求编码器

 * <pre>

 * 数据包格式

 * +——----——+——-----——+——----——+——----——+——-----——+——-----——+

 * | 包头          | 模块号        | 命令号       |  状态码    |  长度          |   数据       |

 * +——----——+——-----——+——----——+——----——+——-----——+——-----——+

 * </pre>

 * 包头4字节

 * 模块号2字节short

 * 命令号2字节short

 * 长度4字节(描述数据部分字节长度)

 */

public class ResponseEncoder extends OneToOneEncoder{

    @Override

    protected Object encode(ChannelHandlerContext context, Channel channel, Object rs) throws Exception {

        Response response = (Response)(rs);

        ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();

        //包头

        buffer.writeInt(ConstantValue.FLAG);

        //module

        buffer.writeShort(response.getModule());

        //cmd

        buffer.writeShort(response.getCmd());

        //状态码

        buffer.writeInt(response.getStateCode());

        //长度

        buffer.writeInt(response.getDataLength());

        //data

        if(response.getData() != null){

            buffer.writeBytes(response.getData());

        }

        return buffer;

    }

}

package com.cn.constant;

public interface ConstantValue {

    /**

     * 包头

     */

    public static final int FLAG = -32523523;

}

package com.cn.model;

/**

 * 客户端请求服务端的对象

 */

public class Request {

    /**

     * 请求模块

     */

    private short module;

    /**

     * 命令号

     */

    private short cmd;

    /**

     * 数据部分

     */

    private byte[] data;

    public short getModule() {

        return module;

    }

    public void setModule(short module) {

        this.module = module;

    }

    public short getCmd() {

        return cmd;

    }

    public void setCmd(short cmd) {

        this.cmd = cmd;

    }

    public byte[] getData() {

        return data;

    }

    public void setData(byte[] data) {

        this.data = data;

    }

    public int getDataLength(){

        if(data == null){

            return 0;

        }

        return data.length;

    }

}

package com.cn.model;

/**

 * 服务端返回给客户端的对象

 */

public class Response {

    /**

     * 请求模块

     */

    private short module;

    /**

     * 命令号

     */

    private short cmd;

    /**

     * 状态码

     */

    private int stateCode;

    /**

     * 数据部分

     */

    private byte[] data;

    public short getModule() {

        return module;

    }

    public void setModule(short module) {

        this.module = module;

    }

    public short getCmd() {

        return cmd;

    }

    public void setCmd(short cmd) {

        this.cmd = cmd;

    }

    public int getStateCode() {

        return stateCode;

    }

    public void setStateCode(int stateCode) {

        this.stateCode = stateCode;

    }

    public byte[] getData() {

        return data;

    }

    public void setData(byte[] data) {

        this.data = data;

    }

    public int getDataLength(){

        if(data == null){

            return 0;

        }

        return data.length;

    }

}

package com.cn.model;

public interface StateCode {

    /**

     * 成功

     */

    public static int SUCCESS  = 0;

    /**

     * 失败

     */

    public static int FAIL  =  1;

}

package com.cn.module.fuben.request;

import com.cn.serial.Serializer;

//FightRequest是模块名

public class FightRequest extends Serializer{

    /**

     * 副本id

     */

    private int fubenId;

    /**

     * 次数

     */

    private int count;

    public int getFubenId() {

        return fubenId;

    }

    public void setFubenId(int fubenId) {

        this.fubenId = fubenId;

    }

    public int getCount() {

        return count;

    }

    public void setCount(int count) {

        this.count = count;

    }

    @Override

    protected void read() {

        this.fubenId = readInt();

        this.count = readInt();

    }

    @Override

    protected void write() {

        writeInt(fubenId);

        writeInt(count);

    }

}

package com.cn.module.fuben.response;

import com.cn.serial.Serializer;

public class FightResponse extends Serializer{

    /**

     * 获取金币

     */

    private int gold;

    public int getGold() {

        return gold;

    }

    public void setGold(int gold) {

        this.gold = gold;

    }

    @Override

    protected void read() {

        this.gold = readInt();

    }

    @Override

    protected void write() {

        writeInt(gold);

    }

}

package com.cn.serial;

import java.nio.ByteOrder;

import org.jboss.netty.buffer.ChannelBuffer;

import org.jboss.netty.buffer.ChannelBuffers;

/**

 * buff工厂

 */

public class BufferFactory {

    public static ByteOrder BYTE_ORDER = ByteOrder.BIG_ENDIAN;

    /**

     * 获取一个buffer

     */

    public static ChannelBuffer getBuffer() {

        ChannelBuffer dynamicBuffer = ChannelBuffers.dynamicBuffer();

        return dynamicBuffer;

    }

    /**

     * 将数据写入buffer

     */

    public static ChannelBuffer getBuffer(byte[] bytes) {

        ChannelBuffer copiedBuffer = ChannelBuffers.copiedBuffer(bytes);

        return copiedBuffer;

    }

}

package com.cn.serial;

import java.nio.charset.Charset;

import java.util.ArrayList;

import java.util.Collection;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import org.jboss.netty.buffer.ChannelBuffer;

/**

 * 自定义序列化接口

 */

public abstract class Serializer {

    public static final Charset CHARSET = Charset.forName("UTF-8");

    protected ChannelBuffer writeBuffer;

    protected ChannelBuffer readBuffer;

    /**

     * 反序列化具体实现

     */

    protected abstract void read();

    /**

     * 序列化具体实现

     */

    protected abstract void write();

    /**

     * 从byte数组获取数据

     * @param bytes    读取的数组

     */

    public Serializer readFromBytes(byte[] bytes) {

        readBuffer = BufferFactory.getBuffer(bytes);

        read();

        readBuffer.clear();

        return this;

    }

    /**

     * 从buff获取数据

     * @param readBuffer

     */

    public void readFromBuffer(ChannelBuffer readBuffer) {

        this.readBuffer = readBuffer;

        read();

    }

    /**

     * 写入本地buff

     * @return

     */

    public ChannelBuffer writeToLocalBuff(){

        writeBuffer = BufferFactory.getBuffer();

        write();

        return writeBuffer;

    }

    /**

     * 写入目标buff

     * @param buffer

     * @return

     */

    public ChannelBuffer writeToTargetBuff(ChannelBuffer buffer){

        writeBuffer = buffer;

        write();

        return writeBuffer;

    }

    /**

     * 返回buffer数组

     *

     * @return

     */

    public byte[] getBytes() {

        writeToLocalBuff();

        byte[] bytes = null;

        if (writeBuffer.writerIndex() == 0) {

            bytes = new byte[0];

        } else {

            bytes = new byte[writeBuffer.writerIndex()];

            writeBuffer.readBytes(bytes);

        }

        writeBuffer.clear();

        return bytes;

    }

    public byte readByte() {

        return readBuffer.readByte();

    }

    public short readShort() {

        return readBuffer.readShort();

    }

    public int readInt() {

        return readBuffer.readInt();

    }

    public long readLong() {

        return readBuffer.readLong();

    }

    public float readFloat() {

        return readBuffer.readFloat();

    }

    public double readDouble() {

        return readBuffer.readDouble();

    }

    public String readString() {

        int size = readBuffer.readShort();

        if (size <= 0) {

            return "";

        }

        byte[] bytes = new byte[size];

        readBuffer.readBytes(bytes);

        return new String(bytes, CHARSET);

    }

    public <T> List<T> readList(Class<T> clz) {

        List<T> list = new ArrayList<>();

        int size = readBuffer.readShort();

        for (int i = 0; i < size; i++) {

            list.add(read(clz));

        }

        return list;

    }

    public <K,V> Map<K,V> readMap(Class<K> keyClz, Class<V> valueClz) {

        Map<K,V> map = new HashMap<>();

        int size = readBuffer.readShort();

        for (int i = 0; i < size; i++) {

            K key = read(keyClz);

            V value = read(valueClz);

            map.put(key, value);

        }

        return map;

    }

    @SuppressWarnings("unchecked")

    public <I> I read(Class<I> clz) {

        Object t = null;

        if ( clz == int.class || clz == Integer.class) {

            t = this.readInt();

        } else if (clz == byte.class || clz == Byte.class){

            t = this.readByte();

        } else if (clz == short.class || clz == Short.class){

            t = this.readShort();

        } else if (clz == long.class || clz == Long.class){

            t = this.readLong();

        } else if (clz == float.class || clz == Float.class){

            t = readFloat();

        } else if (clz == double.class || clz == Double.class){

            t = readDouble();

        } else if (clz == String.class ){

            t = readString();

        } else if (Serializer.class.isAssignableFrom(clz)){

            try {

                byte hasObject = this.readBuffer.readByte();

                if(hasObject == 1){

                    Serializer temp = (Serializer)clz.newInstance();

                    temp.readFromBuffer(this.readBuffer);

                    t = temp;

                }else{

                    t = null;

                }

            } catch (Exception e) {

                e.printStackTrace();

            }

        } else {

            throw new RuntimeException(String.format("不支持类型:[%s]", clz));

        }

        return (I) t;

    }

    public Serializer writeByte(Byte value) {

        writeBuffer.writeByte(value);

        return this;

    }

    public Serializer writeShort(Short value) {

        writeBuffer.writeShort(value);

        return this;

    }

    public Serializer writeInt(Integer value) {

        writeBuffer.writeInt(value);

        return this;

    }

    public Serializer writeLong(Long value) {

        writeBuffer.writeLong(value);

        return this;

    }

    public Serializer writeFloat(Float value) {

        writeBuffer.writeFloat(value);

        return this;

    }

    public Serializer writeDouble(Double value) {

        writeBuffer.writeDouble(value);

        return this;

    }

    public <T> Serializer writeList(List<T> list) {

        if (isEmpty(list)) {

            writeBuffer.writeShort((short) 0);

            return this;

        }

        writeBuffer.writeShort((short) list.size());

        for (T item : list) {

            writeObject(item);

        }

        return this;

    }

    public <K,V> Serializer writeMap(Map<K, V> map) {

        if (isEmpty(map)) {

            writeBuffer.writeShort((short) 0);

            return this;

        }

        writeBuffer.writeShort((short) map.size());

        for (Entry<K, V> entry : map.entrySet()) {

            writeObject(entry.getKey());

            writeObject(entry.getValue());

        }

        return this;

    }

    public Serializer writeString(String value) {

        if (value == null || value.isEmpty()) {

            writeShort((short) 0);

            return this;

        }

        byte data[] = value.getBytes(CHARSET);

        short len = (short) data.length;

        writeBuffer.writeShort(len);

        writeBuffer.writeBytes(data);

        return this;

    }

    public Serializer writeObject(Object object) {

        if(object == null){

            writeByte((byte)0);

        }else{

            if (object instanceof Integer) {

                writeInt((int) object);

                return this;

            }

            if (object instanceof Long) {

                writeLong((long) object);

                return this;

            }

            if (object instanceof Short) {

                writeShort((short) object);

                return this;

            }

            if (object instanceof Byte) {

                writeByte((byte) object);

                return this;

            }

            if (object instanceof String) {

                String value = (String) object;

                writeString(value);

                return this;

            }

            if (object instanceof Serializer) {

                writeByte((byte)1);

                Serializer value = (Serializer) object;

                value.writeToTargetBuff(writeBuffer);

                return this;

            }

            throw new RuntimeException("不可序列化的类型:" + object.getClass());

        }

        return this;

    }

    private <T> boolean isEmpty(Collection<T> c) {

        return c == null || c.size() == 0;

    }

    public <K,V> boolean isEmpty(Map<K,V> c) {

        return c == null || c.size() == 0;

    }

}