[五]RabbitMQ-客户端源码之AMQChannel

欢迎支持笔者新作：《深入理解Kafka:核心设计与实践原理》和《RabbitMQ实战指南》，同时欢迎关注笔者的微信公众号：朱小厮的博客。

欢迎跳转到本文的原文链接：https://honeypps.com/mq/rabbitmq-client-source-code-of-amqchannel/

AMQChannel是一个抽象类，是ChannelN的父类。其中包含唯一的抽象方法：

/*** Protected API - called by nextCommand to check possibly handle an incoming Command before it is returned to the caller of nextCommand. If this method* returns true, the command is considered handled and is not passed back to nextCommand's caller; if it returns false, nextCommand returns the command as* usual. This is used in subclasses to implement handling of Basic.Return and Basic.Deliver messages, as well as Channel.Close and Connection.Close.* @param command the command to handle asynchronously* @return true if we handled the command; otherwise the caller should consider it "unhandled"*/
public abstract boolean processAsync(Command command) throws IOException;

有关processAsync()这个方法的会在介绍ChannelN类的时候详细阐述（[八]RabbitMQ-客户端源码之ChannelN）。

首先来说下AMQChannel的成员变量：

protected final Object _channelMutex = new Object();
/** The connection this channel is associated with. */
private final AMQConnection _connection;
/** This channel's channel number. */
private final int _channelNumber;
/** Command being assembled */
private AMQCommand _command = new AMQCommand();
/** The current outstanding RPC request, if any. (Could become a queue in future.) */
private RpcContinuation _activeRpc = null;
/** Whether transmission of content-bearing methods should be blocked */
public volatile boolean _blockContent = false;

_channelMutex这个是内部用来当对象锁的，没有实际的意义，可忽略
_connection是指AMQConnection这个对象。
_channelNumber是指channel number, 这个应该不用多解释了吧。通道编号为０的代表全局连接中的所有帧，1-65535代表特定通道的帧．
_command是内部处理使用的对象，调用AMQCommand的方法来处理一些东西。
_activeRpc是指当前未处理完的rpc请求（the current outstanding rpc request）。
_blockContent 是在Channel.Flow里用到的，其余情况都是false
在AMQChannel的构造函数中，只有两个参数：AMQConnection connection以及int channelNumber.

AMQChannel中有个handleFrame方法：

/*** Private API - When the Connection receives a Frame for this* channel, it passes it to this method.* @param frame the incoming frame* @throws IOException if an error is encountered*/
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 onehandleCompleteInboundCommand(command);}
}/*** Private API - handle a command which has been assembled* @throws IOException if there's any problem** @param command the incoming command* @throws IOException*/
public void handleCompleteInboundCommand(AMQCommand command) throws IOException {// First, offer the command to the asynchronous-command// handling mechanism, which gets to act as a filter on the// incoming command stream.  If processAsync() returns true,// the command has been dealt with by the filter and so should// not be processed further.  It will return true for// asynchronous commands (deliveries/returns/other events),// and false for commands that should be passed on to some// waiting RPC continuation.if (!processAsync(command)) {// The filter decided not to handle/consume the command,// so it must be some reply to an earlier RPC.nextOutstandingRpc().handleCommand(command);markRpcFinished();}
}

这个在[六]RabbitMQ-客户端源码之AMQCommand有所介绍，主要是用来处理Frame帧的，当调用AMQCommand的handleFrame处理之后返回为true是，即处理完毕时继续调用handleCompleteInboundCommand方法。这其中也牵涉到AMQConnection的MainLoop内部类，具体可以看看：[六]RabbitMQ-客户端源码之AMQCommand。

AMQChannel中有很多方法带有rpc的字样，这来做一个整理。
首先是：

public void enqueueRpc(RpcContinuation k)
{synchronized (_channelMutex) {boolean waitClearedInterruptStatus = false;while (_activeRpc != null) {try {_channelMutex.wait();} catch (InterruptedException e) {waitClearedInterruptStatus = true;}}if (waitClearedInterruptStatus) {Thread.currentThread().interrupt();}_activeRpc = k;}
}

这个方法在AMQConnection.start()方法中有过使用：_channel0.enqueueRpc(conStartBroker)。这个方法就是将参数付给成员变量_activeRpc，至于这个RpcContinuation到底是个什么gui，我们下面再讲。
继续下一个方法：

public boolean isOutstandingRpc()
{synchronized (_channelMutex) {return (_activeRpc != null);}
}

这个方法是判断一下当前的_activeRpc是否为null，为null则为false，否则为true。看方法的名字应该猜出大半。
下面一个方法：

public RpcContinuation nextOutstandingRpc()
{synchronized (_channelMutex) {RpcContinuation result = _activeRpc;_activeRpc = null;_channelMutex.notifyAll();return result;}
}

方法将当前的_activeRpc返回，并置AQMChannel的_activeRpc为null。

接下来几个方法联系性很强：

/*** Protected API - sends a {@link Method} to the broker and waits for the* next in-bound Command from the broker: only for use from* non-connection-MainLoop threads!*/
public AMQCommand rpc(Method m)throws IOException, ShutdownSignalException
{return privateRpc(m);
}public AMQCommand rpc(Method m, int timeout)throws IOException, ShutdownSignalException, TimeoutException {return privateRpc(m, timeout);
}private AMQCommand privateRpc(Method m)throws IOException, ShutdownSignalException
{SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();rpc(m, k);// At this point, the request method has been sent, and we// should wait for the reply to arrive.//// Calling getReply() on the continuation puts us to sleep// until the connection's reader-thread throws the reply over// the fence.return k.getReply();
}private AMQCommand privateRpc(Method m, int timeout)throws IOException, ShutdownSignalException, TimeoutException {SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();rpc(m, k);return k.getReply(timeout);
}public void rpc(Method m, RpcContinuation k)throws IOException
{synchronized (_channelMutex) {ensureIsOpen();quiescingRpc(m, k);}
}public void quiescingRpc(Method m, RpcContinuation k)throws IOException
{synchronized (_channelMutex) {enqueueRpc(k);quiescingTransmit(m);}
}

主要是看最后一个方法——quiescingRpc.这个方法说白就两行代码：
enqueueRpc(k);是将由privateRpc等方法内部创建的SimpleBlockingRpcContinuation对象附给当前的AQMChannel对象的成员变量_activeRpc
关于quiescingTransmit(m)就要接下去看了：

public void quiescingTransmit(Method m) throws IOException {synchronized (_channelMutex) {quiescingTransmit(new AMQCommand(m));}
}
public void quiescingTransmit(AMQCommand c) throws IOException {synchronized (_channelMutex) {if (c.getMethod().hasContent()) {while (_blockContent) {try {_channelMutex.wait();} catch (InterruptedException e) {}// This is to catch a situation when the thread wakes up during// shutdown. Currently, no command that has content is allowed// to send anything in a closing state.ensureIsOpen();}}c.transmit(this);}
}

上面代码只需要看： c.transmit(this);这一句，其余的都是摆设。看到这里，就调用了AMQCommand的transmit方法，这个transmit方法就是讲AMQChannel中封装的内容发给broker，然后等待broker返回，进而通过之前附值的_activeRpc来处理回传的帧。

虽然之前在AMQConnection（[二]RabbitMQ-客户端源码之AMQConnection）中详细讲述了start()方法，但是这里还是要来拿这个来举例这个AMQChannel中的rpc怎么使用
在AMQConnection中有这么一段代码：

Method method = (challenge == null)? new AMQP.Connection.StartOk.Builder().clientProperties(_clientProperties).mechanism(sm.getName()).response(response).build(): new AMQP.Connection.SecureOk.Builder().response(response).build();try {Method serverResponse = _channel0.rpc(method, HANDSHAKE_TIMEOUT/2).getMethod();if (serverResponse instanceof AMQP.Connection.Tune) {connTune = (AMQP.Connection.Tune) serverResponse;} else {challenge = ((AMQP.Connection.Secure) serverResponse).getChallenge();response = sm.handleChallenge(challenge, this.username, this.password);}

客户端将Method封装成Connection.StartOk帧之后等待broker返回Connection.Tune帧。
此时调用了AMQChannel的rpc(Method m, int timeout)方法，其间接调用了AMQChannel的privateRpc(Method, int timeout)方法。代码详情上面已经罗列出来。

注意privateRpc(Method, int timeout)方法的最有一句返回：return k.getReply(timeout);这句代码的意思是SimpleBlockingRpcContinuation对象在等待broker的返回，确切的来说是MainLoop线程处理之后返回，即AMQChannel类中handleCompleteInboundCommand方法的nextOutstandingRpc().handleCommand(command)这行代码。

AQMChannel还有些其他的内容，都是边缘性的东西，这里还剩下个RpcContinuation要着重阐述下的：

public interface RpcContinuation {void handleCommand(AMQCommand command);void handleShutdownSignal(ShutdownSignalException signal);
}public static abstract class BlockingRpcContinuation<T> implements RpcContinuation {public final BlockingValueOrException<T, ShutdownSignalException> _blocker =new BlockingValueOrException<T, ShutdownSignalException>();public void handleCommand(AMQCommand command) {_blocker.setValue(transformReply(command));}public void handleShutdownSignal(ShutdownSignalException signal) {_blocker.setException(signal);}public T getReply() throws ShutdownSignalException{return _blocker.uninterruptibleGetValue();}public T getReply(int timeout)throws ShutdownSignalException, TimeoutException{return _blocker.uninterruptibleGetValue(timeout);}public abstract T transformReply(AMQCommand command);
}public static class SimpleBlockingRpcContinuationextends BlockingRpcContinuation<AMQCommand>
{public AMQCommand transformReply(AMQCommand command) {return command;}
}

RPCContinuation只是一个接口，而BlockingRpcContinuation这个抽象类缺似乎略有门道。而SimpleBlockingRpcContinuation只是将BlockingRpcContinuation中的handleCommand方法便成为：

_blocker.setValue(command);

BlockingRpcContinuation类主要操纵了BlockingValueOrException _blocker这个成员变量。再接下深究BlockingValueOrException其实是继承了BlockingCell，对其做了一下简单的封装。最后来看下BlockingCell是个什么鬼，截取部分代码如下：

public class BlockingCell<T> {private boolean _filled = false;private T _value;public synchronized T get() throws InterruptedException {while (!_filled) {wait();}return _value;}

其实这个就是capacity为1的BlockingQueue，顾美其名曰BlockingCell，绕了大半圈，原来AMQChannel中的_activeRpc就是个这么玩意儿~

附：本系列全集

[Conclusion]RabbitMQ-客户端源码之总结
[一]RabbitMQ-客户端源码之ConnectionFactory
[二]RabbitMQ-客户端源码之AMQConnection
[三]RabbitMQ-客户端源码之ChannelManager
[四]RabbitMQ-客户端源码之Frame
[五]RabbitMQ-客户端源码之AMQChannel
[六]RabbitMQ-客户端源码之AMQCommand
[七]RabbitMQ-客户端源码之AMQPImpl+Method
[八]RabbitMQ-客户端源码之ChannelN
[九]RabbitMQ-客户端源码之Consumer

欢迎跳转到本文的原文链接：https://honeypps.com/mq/rabbitmq-client-source-code-of-amqchannel/

欢迎支持笔者新作：《深入理解Kafka:核心设计与实践原理》和《RabbitMQ实战指南》，同时欢迎关注笔者的微信公众号：朱小厮的博客。

[五]RabbitMQ-客户端源码之AMQChannel相关推荐

RabbitMQ 客户端源码系列 - Channel
前言续上次分享 RabbitMQ 客户端源码系列 - Connection ,继续分享Channel相关的源码分析 (com.rabbitmq:amqp-client:4.8.3) 友情提醒:本次分 ...
RabbitMQ初步到精通-第十章-RabbitMQ之Spring客户端源码
目录第十章-RabbitMQ之Spring客户端源码 1. 前言 2. 客户端消费代码 2.1 消费的实现方式 2.2 消费中注解解释 2.3 推测Spring实现过程 3.MQ消费源码分析 3.1 ...
ZooKeeper客户端源码（零）——客户端API使用
首发CSDN:徐同学呀,原创不易,转载请注明源链接.我是徐同学,用心输出高质量文章,希望对你有所帮助. 本篇源码基于ZooKeeper3.7.0版本. 一.建立连接和会话客户端可以通过创建一个 Zo ...
Libcurl的编译_HTTP/HTTPS客户端源码示例
HTTP/HTTPS客户端源码示例环境: zlib-1.2.8 openssl-1.0.1g curl-7.36 Author: Kagula LastUpdateDate: 2016-05 ...
maven快速入门番外篇——Eclipse下载GitHub上FastDFS-Client客户端源码并转化成maven工程以及打包到本地maven仓库
由于fastdfs-client的jar包目前在中央仓库是没有坐标的,而在项目中要想实现文件的上传和下载就得使用到它,这不禁就让我们头疼,所以为了解决这个问题,我写下了这篇文章,希望对读者能有所帮助. ...
Spring源码深度解析(郝佳)-学习-Spring消息-整合RabbitMQ及源码解析
我们经常在Spring项目中或者Spring Boot项目中使用RabbitMQ,一般使用的时候,己经由前人将配置配置好了,我们只需要写一个注解或者调用一个消息发送或者接收消息的监听器即可,但是底 ...
【微信小程序控制硬件②】开始微信小程序之旅，导入小程序Mqtt客户端源码，实现简单的验证和通讯于服务器.（附带源码）
文章目录一.前言: 二.注册微信小程序: 三.本博文连接和微信物联有何区别: 四.微信小程序`MQTT`客户端源码导入注意事项: 五.下载: 微信物联网生态主要分在微信硬件开发平台与腾讯物联开发平台 ...
boost::asio异步模式的C/S客户端源码实现
异步模式的服务器源码 //g++ -g async_tcp_server.cpp -o async_tcp_server -lboost_system //#include <iostream& ...
zookeeper 客户端_zookeeper进阶-客户端源码详解
流程图先看一下客户端源码的流程图总体流程总体流程开启SendThread线程开启EventThread 总结下面根据源码讲解,大家整合源码和流程图一起看最好,本篇内容比较多建议收藏起来看. ...

[五]RabbitMQ-客户端源码之AMQChannel

附：本系列全集

[五]RabbitMQ-客户端源码之AMQChannel相关推荐

最新文章

热门文章