amqp-client源码解析1：数据格式

amqp-client源码解析1：数据格式

1. Frame

/**
 * 本类代表AMQP中的帧数据
 * 在AMQP协议中，请求/响应一般是由若干个帧（如METHOD + HEADER + BODY...）组成的
 * 类比于HTTP协议的话，这里的帧数据可以代表请求行、请求头和请求体等
 */
public class Frame {
    
    /** 帧类型；有METHOD(1)、HEADER(2)、BODY(3)、HEARTBEAT(8)这4种类型 */
    public final int type;

    /** 该帧所属的通道编号，取值范围为0-65535 */
    public final int channel;

    /** 帧内容（用于入站的帧）；与accumulator互斥（即两者必有一个为null） */
    private final byte[] payload;

    /** 帧内容的输出流（用于出站的帧）；与payload互斥（即两者必有一个为null） */
    private final ByteArrayOutputStream accumulator;

    /** 常量：非帧内容的数据（类型 + 通道编号 + 帧内容长度 + 结束符号）的总大小 */
    private static final int NON_BODY_SIZE = 1 /* type */ + 2 /* channel */ + 4 /* payload size */ + 1 /* end character */;

    /**
     * 当需要给服务器发送帧时，调用本构造方法；后续会把帧内容逐渐添加到this.accumulator中
     */
    public Frame(int type, int channel) {
        this.type = type;
        this.channel = channel;
        this.payload = null;
        this.accumulator = new ByteArrayOutputStream();
    }

    /**
     * 当收到服务器发送过来的一个完整的帧时，调用本构造方法；payload代表解析到的完整的帧内容
     */
    public Frame(int type, int channel, byte[] payload) {
        this.type = type;
        this.channel = channel;
        this.payload = payload;
        this.accumulator = null;
    }

    /**
     * 创建一个BODY类型的帧，并以body[offset:offset+length]作为该帧的帧内容
     */
    public static Frame fromBodyFragment(int channelNumber, byte[] body, int offset, int length) throws IOException {
        Frame frame = new Frame(AMQP.FRAME_BODY, channelNumber);
        DataOutputStream bodyOut = frame.getOutputStream();
        bodyOut.write(body, offset, length);
        return frame;
    }

    /**
     * 从输入流中读取下一个完整的帧
     */
    public static Frame readFrom(DataInputStream is) throws IOException {
        int type;
        int channel;

        // 读取下一个字节作为帧类型；如果超时，说明服务器长时间没有发送数据过来，此时直接返回null
        try {
            type = is.readUnsignedByte();
        } catch (SocketTimeoutException ste) {
            return null; // failed
        }

        // 如果type为'A'，说明服务器返回的数据为"AMQP...."，此时说明客户端和服务器使用的AMQP协议版本不同
        if (type == 'A') {
            
            // 注意这个方法必定会抛出MalformedFrameException，其主要逻辑为：
            // 1. 校验输入流中接下来的3个字符是否为'M'、'Q'、'P'，如果不是，说明响应内容有误
            // 2. 读取接下来的4个字节（服务端的签名），根据签名推断服务端使用的AMQP版本，并构造相应的错误信息
            protocolVersionMismatch(is);
        }

        // 读取接下来的两个字节作为通道编号
        channel = is.readUnsignedShort();
        
        // 读取下一个int类型作为帧内容长度
        int payloadSize = is.readInt();
        
        // 创建相应大小的字节数组，并将输入流中接下来的数据完整地读取到该字节数组中
        byte[] payload = new byte[payloadSize];
        is.readFully(payload);

        // 读取下一个字节；如果该字节不是结束符号，则报错
        int frameEndMarker = is.readUnsignedByte();
        if (frameEndMarker != AMQP.FRAME_END) {
            throw new MalformedFrameException("Bad frame end marker: " + frameEndMarker);
        }

        // 构造Frame实例
        return new Frame(type, channel, payload);
    }

    /**
     * 将本帧的数据写入到输出流中；本方法是readFrom()方法的反向操作
     */
    public void writeTo(DataOutputStream os) throws IOException {
        os.writeByte(type);
        os.writeShort(channel);
        if (accumulator != null) {
            os.writeInt(accumulator.size());
            accumulator.writeTo(os);
        } else {
            os.writeInt(payload.length);
            os.write(payload);
        }
        os.write(AMQP.FRAME_END);
    }

    /**
     * 返回本帧的大小（帧内容大小 + 非帧内容的数据大小）
     */
    public int size() {
        if (accumulator != null) {
            return accumulator.size() + NON_BODY_SIZE;
        } else {
            return payload.length + NON_BODY_SIZE;
        }
    }

    /**
     * 获取帧内容
     */
    public byte[] getPayload() {
        if (payload != null) return payload;
        return accumulator.toByteArray();
    }

    /**
     * 获取帧内容的输出/入流
     */
    public DataInputStream getInputStream() {
        return new DataInputStream(new ByteArrayInputStream(getPayload()));
    }

    public DataOutputStream getOutputStream() {
        return new DataOutputStream(accumulator);
    }
    
    // 省略toString()方法和用于计算大小的静态方法
}

2. Method

2.1. Method接口

/**
 * METHOD帧的帧内容对应的实体类接口
 * 一个Method就代表一种请求/响应类型，并且请求Method和响应Method是成对出现的
 * 比如，假设客户端发起了basic.consume请求，则服务端会返回basic.consume-ok响应
 * 这里的basic.consume就是请求Method，basic.consume-ok就是响应Method
 * 
 * 每个Method都包含一个classId和methodId
 * 反之，一对classId和methodId可以唯一确定一种Method
 */
public interface Method {
    int protocolClassId();
    int protocolMethodId();
    String protocolMethodName();
}

2.2. Method抽象类

/**
 * Method接口的抽象实现类；注意本类类名也是Method
 */
public abstract class Method implements com.rabbitmq.client.Method {

    /**
     * 这种Method对应的请求（或响应）是否包含请求头和请求体（或响应头和响应体）
     * 比如，basic.get请求没有请求头和请求体，basic.get-ok响应有响应头和响应体（即消息内容）
     */
    public abstract boolean hasContent();

    /**
     * 支持访问者模式
     * MethodVisitor接口有若干个重载的visit()方法
     * 假设本对象是Basic.Get类型，则本方法会调用MethodVisitor实例的Object visit(Basic.Get x)方法，即return visitor.visit(this);
     */
    public abstract Object visit(MethodVisitor visitor) throws IOException;

    /**
     * 将本Method的额外参数信息写入到输出流中
     */
    public abstract void writeArgumentsTo(MethodArgumentWriter writer) throws IOException;

    /**
     * 将本Method转成对应的METHOD帧
     */
    public Frame toFrame(int channelNumber) throws IOException {
        
        // 创建METHOD帧
        Frame frame = new Frame(AMQP.FRAME_METHOD, channelNumber);
        
        // 将本Method的信息作为帧内容写入到输出流中；包括本Method的classId、methodId和额外参数信息
        DataOutputStream bodyOut = frame.getOutputStream();
        bodyOut.writeShort(protocolClassId());
        bodyOut.writeShort(protocolMethodId());
        MethodArgumentWriter argWriter = new MethodArgumentWriter(new ValueWriter(bodyOut));
        writeArgumentsTo(argWriter);
        argWriter.flush();
        return frame;
    }
}

2.3. Ack

/**
 * 本类是AMQImpl#Basic类的静态内部类，继承自上面的抽象Method类，并实现了Ack接口（继承自Method接口）；注意本类类名也是Ack
 */
public static class Ack extends Method implements com.rabbitmq.client.AMQP.Basic.Ack {
    public static final int INDEX = 80;

    /**
     * basic.ack方式有deliveryTag和multiple这两个额外参数
     * 这和Channel接口的basicAck(long deliveryTag, boolean multiple)方法是完全对应的
     */
    private final long deliveryTag;
    private final boolean multiple;
    
    // 省略上面两个字段的get方法和全参构造器

    /**
     * 根据输入流来构造Ack实例；这里会从输入流中读取到deliveryTag和multiple信息并用来构造本类实例
     */
    public Ack(MethodArgumentReader rdr) throws IOException { this(rdr.readLonglong(), rdr.readBit()); }

    /**
     * 实现抽象方法
     */
    public int protocolClassId() { return 60; }
    public int protocolMethodId() { return 80; }
    public String protocolMethodName() { return "basic.ack"; }
    public boolean hasContent() { return false; }
    public Object visit(MethodVisitor visitor) throws IOException { return visitor.visit(this); }

    /**
     * 将本Method的额外参数信息写入到输出流中
     */
    public void writeArgumentsTo(MethodArgumentWriter writer) throws IOException {
        writer.writeLonglong(this.deliveryTag);
        writer.writeBit(this.multiple);
    }
}

3. ContentHeader

3.1. ContentHeader

/**
 * HEADER帧的帧内容对应的实体类接口
 * 
 * 每个Header都包含一个classId
 * 反之，一个classId可以唯一确定一种Header
 * 目前Header有且只有一种，其classId = 60、className = "basic"
 */
public interface ContentHeader extends Cloneable {
    int getClassId();
    String getClassName();
    void appendPropertyDebugStringTo(StringBuilder buffer);  // 该方法用于debug，忽略即可
}

3.2. AMQContentHeader

/**
 * 本类是ContentHeader接口抽象实现类
 */
public abstract class AMQContentHeader implements ContentHeader {
    
    /** 请求体/响应体的总大小（默认为0）；省略其get方法 */
    private long bodySize;
    
    protected AMQContentHeader() {
        this.bodySize = 0;
    }

    /**
     * 根据输入流来构造AMQContentHeader实例
     * 这里会先读取掉前2个占位字节，然后读取接下来的8个字节作为bodySize
     */
    protected AMQContentHeader(DataInputStream in) throws IOException {
        in.readShort(); // weight not currently used
        this.bodySize = in.readLong();
    }

    /**
     * 将本Header的数据写入到输出流中
     * 先写入2个占位字节，再写入bodySize，最后写入一些额外的头信息
     */
    private void writeTo(DataOutputStream out, long bodySize) throws IOException {
        out.writeShort(0); // weight - not currently used
        out.writeLong(bodySize);
        writePropertiesTo(new ContentHeaderPropertyWriter(out));
    }

    /**
     * 抽象方法：将本Header的额外的头信息写入到输出流中
     */
    public abstract void writePropertiesTo(ContentHeaderPropertyWriter writer) throws IOException;

    /**
     * 将本Header转成对应的HEADER帧
     */
    public Frame toFrame(int channelNumber, long bodySize) throws IOException {
        Frame frame = new Frame(AMQP.FRAME_HEADER, channelNumber);
        DataOutputStream bodyOut = frame.getOutputStream();
        bodyOut.writeShort(getClassId());
        writeTo(bodyOut, bodySize);
        return frame;
    }
}

4. Command

4.1. Command

/**
 * 一个Command就代表一次请求/响应的数据，包含Method、ContentHeader和具体的请求体/响应体数据
 */
public interface Command {
    Method getMethod();
    ContentHeader getContentHeader();
    byte[] getContentBody();
}

4.2. CommandAssembler

/**
 * 命令组装器
 * 本组件负责将若干个帧组装成一个完整的Command实例
 * 本类所有的公开方法都是同步的，因此是线程安全的
 */
final class CommandAssembler {
    private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];

    /** 本组件内部通过状态机来实现对命令的组装；这里定义了4种状态 */
    private enum CAState {
        EXPECTING_METHOD,         // 下一个帧必须是METHOD帧（初始状态）
        EXPECTING_CONTENT_HEADER, // 下一个帧必须是HEADER帧
        EXPECTING_CONTENT_BODY,   // 下一个帧必须是BODY帧
        COMPLETE                  // 已解析完成
    }
    
    /** 当前状态 */
    private CAState state;

    /** 解析到的Method */
    private Method method;
    
    /** 解析到的Header */
    private AMQContentHeader contentHeader;

    /** 请求体/响应体内容可能比较大，此时需要使用多个BODY帧来传输数据；这个列表就用于存放所有BODY帧的内容 */
    private final List<byte[]> bodyN;
    
    /** 请求体/响应体的总大小 */
    private int bodyLength;

    /** 剩余的请求体/响应体的大小 */
    private long remainingBodyBytes;

    /**
     * 当解析到一个BODY帧内容后，调用本方法将其保存起来
     */
    private void appendBodyFragment(byte[] fragment) {
        if (fragment == null || fragment.length == 0) return;
        bodyN.add(fragment);
        bodyLength += fragment.length;
    }

    /**
     * 构造方法；如果之前已经解析到了Method、Header或Body，则可以将它们通过构造参数传进来，本组件会接着往下解析
     */
    public CommandAssembler(Method method, AMQContentHeader contentHeader, byte[] body) {
        this.method = method;
        this.contentHeader = contentHeader;
        this.bodyN = new ArrayList<byte[]>(2);
        this.bodyLength = 0;
        this.remainingBodyBytes = 0;
        appendBodyFragment(body);
        
        // 如果Method为空，则需要从头开始解析
        if (method == null) {
            this.state = CAState.EXPECTING_METHOD;
        
        // 否则，如果Header为空，则根据该Method是否有内容来推断当前状态；如果有，则下一步需要解析Header，否则直接解析完成
        } else if (contentHeader == null) {
            this.state = method.hasContent() ? CAState.EXPECTING_CONTENT_HEADER : CAState.COMPLETE;
        
        // 否则，从Header中读取到请求体/响应体的总大小，并计算出剩余的请求体/响应体的大小，并更新当前状态
        } else {
            this.remainingBodyBytes = contentHeader.getBodySize() - this.bodyLength;
            updateContentBodyState();
        }
    }

    /** 
     * 根据this.remainingBodyBytes来设置当前状态：如果大于0，则还需要再读取BODY帧，否则解析完成
     */
    private void updateContentBodyState() {
        this.state = (this.remainingBodyBytes > 0) ? CAState.EXPECTING_CONTENT_BODY : CAState.COMPLETE;
    }

    // 省略简单的get方法

    /**
     * 核心方法；当接收到一个帧后，调用本方法来处理该帧
     * 这里会根据当前状态来调用对应的方法来处理该帧
     * 返回值代表是否解析完成（即能否组装成一个完成的Command）
     */
    public synchronized boolean handleFrame(Frame f) throws IOException {
        switch (this.state) {
            case EXPECTING_METHOD:          consumeMethodFrame(f); break;
            case EXPECTING_CONTENT_HEADER:  consumeHeaderFrame(f); break;
            case EXPECTING_CONTENT_BODY:    consumeBodyFrame(f);   break;
            default:                        throw new IllegalStateException("Bad Command State " + this.state);
        }
        return isComplete();
    }

    /**
     * 处理METHOD帧，主要逻辑：
     * 1. 先从该帧的输入流中读取classId和methodId，确定该Method的具体类型（如basic.ack）
     * 2. 继续读取该Method的额外参数信息，最终构造出对应的实体类实例（如AMQImpl#Basic#Ack类实例）并赋值给this.method
     * 3. 根据该Method是否有内容来推断当前状态；如果有，则下一步需要解析Header，否则直接解析完成
     */
    private void consumeMethodFrame(Frame f) throws IOException {
        if (f.type == AMQP.FRAME_METHOD) {
            this.method = AMQImpl.readMethodFrom(f.getInputStream());
            this.state = this.method.hasContent() ? CAState.EXPECTING_CONTENT_HEADER : CAState.COMPLETE;
        } else {
            throw new UnexpectedFrameError(f, AMQP.FRAME_METHOD);
        }
    }

    /**
     * 处理HEADER帧，主要逻辑在上文中基本都有提及（或者有类似的操作）
     */
    private void consumeHeaderFrame(Frame f) throws IOException {
        if (f.type == AMQP.FRAME_HEADER) {
            this.contentHeader = AMQImpl.readContentHeaderFrom(f.getInputStream());
            this.remainingBodyBytes = this.contentHeader.getBodySize();
            updateContentBodyState();
        } else {
            throw new UnexpectedFrameError(f, AMQP.FRAME_HEADER);
        }
    }

    /**
     * 处理BODY帧
     */
    private void consumeBodyFrame(Frame f) {
        if (f.type == AMQP.FRAME_BODY) {
            byte[] fragment = f.getPayload();
            this.remainingBodyBytes -= fragment.length;
            updateContentBodyState();
            if (this.remainingBodyBytes < 0) {
                throw new UnsupportedOperationException("%%%%%% FIXME unimplemented");
            }
            appendBodyFragment(fragment);
        } else {
            throw new UnexpectedFrameError(f, AMQP.FRAME_BODY);
        }
    }

    /**
     * 将this.bodyN列表中的所有字节数组合并成一个字节数组，并返回该字节数组
     */
    private byte[] coalesceContentBody() {
        if (this.bodyLength == 0) return EMPTY_BYTE_ARRAY;
        if (this.bodyN.size() == 1) return this.bodyN.get(0);

        byte[] body = new byte[bodyLength];
        int offset = 0;
        for (byte[] fragment : this.bodyN) {
            System.arraycopy(fragment, 0, body, offset, fragment.length);
            offset += fragment.length;
        }
        this.bodyN.clear();
        this.bodyN.add(body);
        return body;
    }
}

4.3. AMQCommand

/**
 * 本类是Command接口的唯一实现类
 * 本类的所有功能基本都是靠底层的CommandAssembler组件完成的
 */
public class AMQCommand implements Command {

    /** 空帧的大小，与Frame#NON_BODY_SIZE相同 */
    public static final int EMPTY_FRAME_SIZE = 8;

    /** 底层的命令组装器 */
    private final CommandAssembler assembler;

    /**
     * 构造方法，这里的构造参数全部用于初始化底层的CommandAssembler组件
     * 本类的其它构造方法都是在调用本方法，因此省略掉
     */
    public AMQCommand(com.rabbitmq.client.Method method, AMQContentHeader contentHeader, byte[] body) {
        this.assembler = new CommandAssembler((Method) method, contentHeader, body);
    }

    @Override
    public Method getMethod() {
        return this.assembler.getMethod();
    }

    @Override
    public AMQContentHeader getContentHeader() {
        return this.assembler.getContentHeader();
    }

    @Override
    public byte[] getContentBody() {
        return this.assembler.getContentBody();
    }

    public boolean handleFrame(Frame f) throws IOException {
        return this.assembler.handleFrame(f);
    }

    /**
     * 将本命令（即请求）通过指定的通道发送给服务端
     * 这里本质上是在往通道底层的连接写入帧数据，并且这些帧的通道编号都为channel.getChannelNumber()
     */
    public void transmit(AMQChannel channel) throws IOException {
        int channelNumber = channel.getChannelNumber();
        AMQConnection connection = channel.getConnection();

        // 对CommandAssembler组件进行加锁
        synchronized (assembler) {
            Method m = this.assembler.getMethod();
            
            // 如果该Method含有请求头和请求体，则需要写入三部分内容
            if (m.hasContent()) {
                
                // 获取到请求体的完整数据，并将请求头转成Frame实例
                byte[] body = this.assembler.getContentBody();
                Frame headerFrame = this.assembler.getContentHeader().toFrame(channelNumber, body.length);

                // 获取到帧的最大大小（0代表不限制），并计算BODY帧内容的最大大小（frameMax - EMPTY_FRAME_SIZE）
                int frameMax = connection.getFrameMax();
                boolean cappedFrameMax = frameMax > 0;
                int bodyPayloadMax = cappedFrameMax ? frameMax - EMPTY_FRAME_SIZE : body.length;

                // 如果请求头的帧大小超过了帧的最大大小，则报错
                if (cappedFrameMax && headerFrame.size() > frameMax) {
                    String msg = String.format("Content headers exceeded max frame size: %d > %d", headerFrame.size(), frameMax);
                    throw new IllegalArgumentException(msg);
                }
                
                // 写入METHOD帧和HEADER帧
                connection.writeFrame(m.toFrame(channelNumber));
                connection.writeFrame(headerFrame);

                // 分批次写入BODY帧，确保每个BODY帧不会超过最大大小
                for (int offset = 0; offset < body.length; offset += bodyPayloadMax) {
                    int remaining = body.length - offset;
                    int fragmentLength = (remaining < bodyPayloadMax) ? remaining : bodyPayloadMax;
                    Frame frame = Frame.fromBodyFragment(channelNumber, body, offset, fragmentLength);
                    connection.writeFrame(frame);
                }
            
            // 否则，直接发送METHOD帧即可
            } else {
                connection.writeFrame(m.toFrame(channelNumber));
            }
        }
        connection.flush();
    }
    
    // 省略其它不重要的方法
}