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中提到的参考文档
