MQ!Rabbit-client command

在 RabbitMq-clien [建立链接的源码]

在看mainLoop线程的时候我们留下了一个下尾巴，来还债。

入口 this._frameHandler.initialize(this);

下面是MainLoop的代码

public void run() {
    try {
        while (_running) {
            Frame frame = _frameHandler.readFrame();
            // 处理帧
            readFrame(frame);
        }
    } catch (Throwable ex) {
        handleFailure(ex);
    } finally {
        doFinalShutdown();
    }
}

// readFrame(frame)内容
private void readFrame(Frame frame) throws IOException {
    if (frame != null) {
        _missedHeartbeats = 0;
        // 如果是心跳帧则忽略
        if (frame.type == AMQP.FRAME_HEARTBEAT) {
            // Ignore it: we've already just reset the heartbeat counter.
        } else {
            if (frame.channel == 0) { // the special channel
                // 特殊帧
                _channel0.handleFrame(frame);
            } else {
                if (isOpen()) {
                    // If we're still _running, but not isOpen(), then we
                    // must be quiescing, which means any inbound frames
                    // for non-zero channels (and any inbound commands on
                    // channel zero that aren't Connection.CloseOk) must
                    // be discarded.
                    ChannelManager cm = _channelManager;
                    if (cm != null) {
                        ChannelN channel;
                        try {
                            channel = cm.getChannel(frame.channel);
                        } catch(UnknownChannelException e) {
                            // this can happen if channel has been closed,
                            // but there was e.g. an in-flight delivery.
                            // just ignoring the frame to avoid closing the whole connection
                            LOGGER.info("Received a frame on an unknown channel, ignoring it");
                            return;
                        }
                        // 处理
                        channel.handleFrame(frame);
                    }
                }
            }
        }
    } else {
        // Socket timeout waiting for a frame.
        // Maybe missed heartbeat.
        handleSocketTimeout();
    }
}

// channel.handleFrame  后面在进行分析
public void handleFrame(Frame frame) throws IOException {
    AMQCommand command = _command;
    if (command.handleFrame(frame)) { // a complete command has rolled off the assembly line
        _command = new AMQCommand(); // prepare for the next one
        handleCompleteInboundCommand(command);
    }
}

我不会告诉你，上面的都是为了码字数的

AMQCommand 实现 Command

Command

package com.rabbitmq.client;

/**
 * Interface to a container for an AMQP method-and-arguments, with optional content header and body.
 */
public interface Command {

    /**
     * @return the command's method.
     */
    Method getMethod();

    /**
     * @return the Command's {@link ContentHeader}, or null if none
     */
    ContentHeader getContentHeader();

    /**
     * @return the Command's content body, or null if none
     */
    byte[] getContentBody();
}

从这个类可以看到Commamnd中有 method contendHeader 和一个字节数组 contentBody

public interface ContentHeader extends Cloneable {    
    public abstract int getClassId();    
    public abstract String getClassName();
    public void appendPropertyDebugStringTo(StringBuilder buffer);
}

command处理

先不废话，继续看下MainLoop里面是如何处理的

AMQCommand -> command.handleFrame(frame)

private final CommandAssembler assembler;

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

这里处理帧的操作委托给了CommandAssembler继续读。

/** Current state, used to decide how to handle each incoming frame. */
private enum CAState {
    // 准备处理Method
    EXPECTING_METHOD, 
    // 处理Content header
    EXPECTING_CONTENT_HEADER, 
    // 处理Content body
    EXPECTING_CONTENT_BODY, 
    // 完成
    COMPLETE
}
private CAState state;

/** The method for this command */
private Method method;

/** The content header for this command */
private AMQContentHeader contentHeader;
// AMQP帧
private final List<byte[]> bodyN;
// 消息体大小
private int bodyLength;
/** No bytes of content body not yet accumulated */
private long remainingBodyBytes;

看下CommandAssembler的handleFrame()方法(这里注意方法中的修饰词synchronized,其实仔细查看的话，这个类对外暴漏的方法都是有这个关键字的，就表示这个类是线程安全的)

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();
}

对具体的处理有兴趣的可以自行读下里面的方法。

这里要特别说明，CommandAssembler.handleFrame()只是对一个帧来进行处理，有的指令可能包含多个帧。

即在MainLoop的run方法中,使用了while循环来处理每一帧

while (_running) {
    // 这里面用到了上一篇文章里面写的Frame类的read方法，调用前使用了synchronized来保证线程安全
    Frame frame = _frameHandler.readFrame();
    readFrame(frame);
}

AMQCommand中还有一个核心方法：将Command转换为多个Frame并且发送。这个类具体是在何时调用后续在进行学习

public void transmit(AMQChannel channel) throws IOException {
    int channelNumber = channel.getChannelNumber();
    AMQConnection connection = channel.getConnection();

    synchronized (assembler) {
        Method m = this.assembler.getMethod();
        connection.writeFrame(m.toFrame(channelNumber));
        if (m.hasContent()) {
            byte[] body = this.assembler.getContentBody();

            connection.writeFrame(this.assembler.getContentHeader()
                                  .toFrame(channelNumber, body.length));

            int frameMax = connection.getFrameMax();
            //如果body长度超过client与server协商出的最大帧长度，将分多个Frame发送
            int bodyPayloadMax = (frameMax == 0) ? body.length : frameMax
                - EMPTY_FRAME_SIZE;

            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);
            }
        }
    }

    connection.flush();
}

如果想更深入理解可以参考上一篇FrameHandler中提到的参考文档