[五]RabbitMQ-客户端源码之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 one
        handleCompleteInboundCommand(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 SimpleBlockingRpcContinuation
    extends 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就是个这么玩意儿~

附：本系列全集

  1. [Conclusion]RabbitMQ-客户端源码之总结
  2. [一]RabbitMQ-客户端源码之ConnectionFactory
  3. [二]RabbitMQ-客户端源码之AMQConnection
  4. [三]RabbitMQ-客户端源码之ChannelManager
  5. [四]RabbitMQ-客户端源码之Frame
  6. [五]RabbitMQ-客户端源码之AMQChannel
  7. [六]RabbitMQ-客户端源码之AMQCommand
  8. [七]RabbitMQ-客户端源码之AMQPImpl+Method
  9. [八]RabbitMQ-客户端源码之ChannelN
  10. [九]RabbitMQ-客户端源码之Consumer
时间： 2024-10-21 13:26:35

[五]RabbitMQ-客户端源码之AMQChannel的相关文章

memcached客户端源码分析

转载:memcached客户端源码分析 memcached的Java客户端有好几种,http://code.google.com/p/memcached/wiki/Clients 罗列了以下几种 Html代码   spymemcached          * http://www.couchbase.org/code/couchbase/java             o An improved Java API maintained by Matt Ingenthron and other

急求百度手环手机客户端源码

问题描述 急求百度手环手机客户端源码 50C dulife手环的安卓手机客户端的源码或者手机端测试的源码百度手环Android手机客户端源码 解决方案 这种源码怎么可能给你,都是商业机密.去研究研究原理.技术还是正途.

开发一个Linux调试器（五）：源码和信号

在上一部分我们学习了关于 DWARF 的信息,以及它如何被用于读取变量和将被执行的机器码与我们的高级语言的源码联系起来.在这一部分,我们将进入实践,实现一些我们调试器后面会使用的 DWARF 原语.我们也会利用这个机会,使我们的调试器可以在命中一个断点时打印出当前的源码上下文. 系列文章索引 随着后面文章的发布,这些链接会逐渐生效. 准备环境 断点 寄存器和内存 Elves 和 dwarves 源码和信号 源码级逐步执行 源码级断点 调用栈展开 读取变量 下一步 设置我们的 DWARF 解析器

求java开发的FTP服务器与客户端源码，必有重谢！

问题描述 要求界面和谐,有权限控制,各位大大手头上有的请帮个忙吧!

android-dulife安卓手机客户端的源码,急求

问题描述 dulife安卓手机客户端的源码,急求 dulife手环的安卓手机客户端的源码,或者手机端测试的源码,百度手环Android手机客户端源码

[一]RabbitMQ-客户端源码之ConnectionFactory

首先看一段amqp-client发送端的示例代码(展示出主要部分): ConnectionFactory factory = new ConnectionFactory(); factory.setHost(ip); factory.setPort(5672); factory.setUsername("root"); factory.setPassword("root"); Connection connection = factory.newConnection

[Conclusion]RabbitMQ-客户端源码之总结

RabbitMQ遵从的是AMQP协议,其broker端代码采用erlang编写,对于没有接触过erlang的同学(包括博主我)来说,想要了解其中的奥秘实在是不容易,大多只能从网上"搜刮"点散碎的知识点来充实一下.但是这样是不能究其然,更不能究其所以然.博主这里翻阅了amqp-client的java客户端的源码,通过其来学习下AMQP协议,进而更深刻的了解RabbitMQ. 注:如无特殊说明,本系列的文章采用的amqp-client版本均为3.5.3. 本系列的文章主要是来阐述客户端与b

[二]RabbitMQ-客户端源码之AMQConnection

上一篇文章([一]RabbitMQ-客户端源码之ConnectionFactory)中阐述了conn.start()方法完成之后客户端就已经和broker建立了正常的连接,而这个Connection的关键就在于这个start()方法之内,下面我们来慢慢分析. 首先来看看start()方法的源码,这个方法有点长,这里拆开来一一分析,首先是注释: /** * Start up the connection, including the MainLoop thread. * Sends the pro

[七]RabbitMQ-客户端源码之AMQPImpl+Method

AMQPImpl类包括AMQP接口(public class AMQImpl implements AMQP)主要囊括了AMQP协议中的通信帧的类别. 这里以Connection.Start帧做一个例子. public static class Connection { public static final int INDEX = 10; public static class Start extends Method implements com.rabbitmq.client.AMQP.C