RabbitMQ初步到精通-第八章-Java-AMQP-Client源码分析

创始人

2024-02-04 22:24:01

0次

第八章-Java-AMQP-Client源码分析

1、背景

1.1 客户端介绍

1.2 看源码好处

1.3 如何看源码

2、生产者

3、消费者监听

4、创建连接

5、消费者消费

6. 总结：

第八章-Java-AMQP-Client源码分析

1、背景

1.1 客户端介绍

通过前面几章的学习，大家对rabbitmq 的基本使用应该ok了，但有的同学可能仍然不满足先去看看Rabbitmq如何实现的，由于rabbitmq 使用Erlang实现的，大家可以自行研究。

看不了mq的实现，可以看下他的客户端的实现。客户端也有多种语言的实现，我们以java的amqp-client来看。

com.rabbitmqamqp-client5.7.3

1.2 看源码好处

另外有些同学会有疑问，能用不就行了，看这源码有啥用呢。

首先，看源码能满足我们的好奇心，做到知其然又知其所以然。

其次，在实际运用的过程中，出现的一些问题，需要靠看源码来分析解决。例如，我们之前发现mq的消费很慢，但是消费者处理速度和生产者处理的速度都很快，所以想从mq的客户端看看，有没有什么瓶颈。

再有，看源码能提升我们的编码能力，学到很多优秀的编码习惯，算法，架构等等。既然这些中间件能开源出来，被广泛使用，肯定有他优秀的地方，开阔自己视野，站在巨人的肩膀上看世界。

等等...

1.3 如何看源码

有的同学可能认为，直接拔开就看呗，一个类一个类的，一个方法一个方法的看。从上往下。不可否认这是一种看法，但这不太适合初期刚看代码的时机，会搞的很懵，直接失去看源码的兴趣。

总结几个小方法：

1、可以把源码下载到本地，部署起来，一定要能跑起来。另外也可以省事些，在IDE里面点进Jar包，下载源码，直接从Jar包里看。

2、按图索骥，看的时候一定不是按一个网来看的，而是专注的一个点，从这个点进去，一步一步跟随到源码中，串成一条线，最后很多的线就会组成一个网，是逐步按照点、线、面的方式来。

3、开启Debug，直接读有时候会绕进去，找不到绕出来的方向，因为源码中各个类的实现关联都很多，不如直接按Debug模式，跟随着读进去。这里面有一个很重要的点，-有的源码中是靠新启线程实现的，所以记得Debug断点的时候，要使用Thread模式哦。

4、抓大放小，读源码我们有时候需要一些不求甚解，需要一些模棱两可，我们无法搞懂所有的东西，特别是刚读的时候，但我们一定要清楚哪些是主线，哪些是边角料。把主线搞清楚即可。

5、Again and Again, 源码不可能一遍就让你读懂，都熟悉。这是需要不断的重复的一个过程，一遍不懂，就再来一遍，十遍不行就二十遍，三十遍，每一遍都会有新的收获。

6、坚持，不轻易放弃。

后续我们看源码的这几个点，也是按照我们之前讲过的 RabbitMQ的 Simple模式，最简单的案例，涉及到的，追踪到源码中去分析。

2、生产者

生产者，代码很简单，追进去，也比较清晰。

业务代码：

channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());

这就是生产者通过channel 发布了一条消息给默认的 Exchange，并指定了队列的名称。

好，追进去，一直追到 ChannelIN 的 basicPublish方法：

    /** Public API - {@inheritDoc} */@Overridepublic void basicPublish(String exchange, String routingKey,boolean mandatory, boolean immediate,BasicProperties props, byte[] body)throws IOException{if (nextPublishSeqNo > 0) {unconfirmedSet.add(getNextPublishSeqNo());nextPublishSeqNo++;}if (props == null) {props = MessageProperties.MINIMAL_BASIC;}//组装 AMQCommand对象，后续进行网络传输 // 拼装了 交换机，路由键，消息等内容AMQCommand command = new AMQCommand(new Basic.Publish.Builder().exchange(exchange).routingKey(routingKey).mandatory(mandatory).immediate(immediate).build(), props, body);try {// 核心发送方法transmit(command);} catch (IOException e) {metricsCollector.basicPublishFailure(this, e);throw e;}metricsCollector.basicPublish(this);}

继续追transmit方法，追至 AMQCommand.transmit 方法即可，中间其他的方法可以略过

这里面的内容也没啥太多关注的，就是拿到Connection去写信息，最后Flush过去。

public void transmit(AMQChannel channel) throws IOException {int channelNumber = channel.getChannelNumber();AMQConnection connection = channel.getConnection();synchronized (assembler) {Method m = this.assembler.getMethod();if (m.hasContent()) {byte[] body = this.assembler.getContentBody();Frame headerFrame = this.assembler.getContentHeader().toFrame(channelNumber, body.length);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);}connection.writeFrame(m.toFrame(channelNumber));connection.writeFrame(headerFrame);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);}} else {connection.writeFrame(m.toFrame(channelNumber));}}connection.flush();}

至此就把，消息推送到了 MQ Broker。

大家用抓包工具抓下会看的更清晰：

这一次的发送有3个AMQP协议的内容

第一个，Method

这些都是我们代码中的参数，完全匹配：

第二个：头信息

第三个：消息内容：

3、消费者监听

业务代码是要开启一个监听然后将此监听发送到MQ中

//4.开启监听QueueDefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("小明洗澡用水: " + new String(body, "UTF-8"));}};/*** 参数1：queue 指定消费哪个队列* 参数1：deliverCallback 指定是否ACK（true:收到消息会立即告诉RabbiMQ，false：手动告诉）* 参数1：cancelCallback 指定消费回调*/channel.basicConsume(QUEUE_NAME, true, consumer);

我们从basicConsume跟进去：一直到ChannelN 的 basicConsume 中去：

   /** Public API - {@inheritDoc} */@Overridepublic String basicConsume(String queue, final boolean autoAck, String consumerTag,boolean noLocal, boolean exclusive, Map arguments,final Consumer callback)throws IOException{// 拼装 Method 对象，Basic.Consume 后续传输使用final Method m = new Basic.Consume.Builder().queue(queue).consumerTag(consumerTag).noLocal(noLocal).noAck(autoAck).exclusive(exclusive).arguments(arguments).build();// 声明监听对象 为后续 传输至MQ，MQ返回消息接收使用BlockingRpcContinuation k = new BlockingRpcContinuation(m) {@Overridepublic String transformReply(AMQCommand replyCommand) {String actualConsumerTag = ((Basic.ConsumeOk) replyCommand.getMethod()).getConsumerTag();_consumers.put(actualConsumerTag, callback);// need to register consumer in stats before it actually starts consumingmetricsCollector.basicConsume(ChannelN.this, actualConsumerTag, autoAck);dispatcher.handleConsumeOk(callback, actualConsumerTag);return actualConsumerTag;}};// 核心调用 传输信息rpc(m, k);try {if(_rpcTimeout == NO_RPC_TIMEOUT) {return k.getReply();} else {try {return k.getReply(_rpcTimeout);} catch (TimeoutException e) {throw wrapTimeoutException(m, e);}}} catch(ShutdownSignalException ex) {throw wrap(ex);}}

然后经过后面的方法，还会上面的Method对象包装一层成为 AMQCommand，最后又调用到了和生产者调用一致的部分：不再赘述了。

AMQPCommand
public void transmit(AMQChannel channel) throws IOException {

到此为止，就会出现疑问了，我们把消费的信息推送给MQ了，啥时候消费，啥时候调用我们自定义的监听的消费方法呢？这里相当于一个异步了，断层了。

这个就得往后看了，实际上是后续MQ得知有消费者注册到queue上之后，就会推送给消费者消息，消费者再去获取这个消息。先不急。

看下网络数据包的情况：

这时候我们先产生一个交互，先告诉MQ说，我是个消费者，想要消费SolarWaterHeater 这个队列的消息了。MQ如果告诉他，可以。后续MQ再推送消息过来。

协议信息内容：

4、创建连接

我们看下创建连接的部分：因为无论生产者和消费者与MQ交互都得首先创建连接，而且创建连接里面还干了一件重要的事，来解决刚才上面提到的如何消费MQ推送过来的消息的问题。

业务创建连接代码：

  Connection connection = null;//创建Connection工厂ConnectionFactory factory = new ConnectionFactory();factory.setVirtualHost("my-test-virtual");factory.setPassword("test");factory.setUsername("test");factory.setHost("127.0.0.1");factory.setPort(5672);//拿到连接try {connection = factory.newConnection();

跟进newConnection : 来到 ConnectionFactory .newConnection

 public Connection newConnection(ExecutorService executor, AddressResolver addressResolver, String clientProvidedName)throws IOException, TimeoutException {if(this.metricsCollector == null) {this.metricsCollector = new NoOpMetricsCollector();}// make sure we respect the provided thread factoryFrameHandlerFactory fhFactory = createFrameHandlerFactory();ConnectionParams params = params(executor);// set client-provided via a client propertyif (clientProvidedName != null) {Map properties = new HashMap(params.getClientProperties());properties.put("connection_name", clientProvidedName);params.setClientProperties(properties);}if (isAutomaticRecoveryEnabled()) {// see com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnectionFactory#newConnectionAutorecoveringConnection conn = new AutorecoveringConnection(params, fhFactory, addressResolver, metricsCollector);conn.init();return conn;} else {List addrs = addressResolver.getAddresses();Exception lastException = null;for (Address addr : addrs) {try {// 创建 FrameHandlerFrameHandler handler = fhFactory.create(addr, clientProvidedName);// 组装AMQConnection 对象AMQConnection conn = createConnection(params, handler, metricsCollector);// 核心启动conn.start();this.metricsCollector.newConnection(conn);return conn;} catch (IOException e) {lastException = e;} catch (TimeoutException te) {lastException = te;}}if (lastException != null) {if (lastException instanceof IOException) {throw (IOException) lastException;} else if (lastException instanceof TimeoutException) {throw (TimeoutException) lastException;}}throw new IOException("failed to connect");}}

所做的一切，拿配置，地址，拼接FrameHandler 都是为了组装 AMQConnection 对象，组装对象完成后即需要，conn.start(); 启动连接。继续往下跟：

  /*** Start up the connection, including the MainLoop thread.* Sends the protocol* version negotiation header, and runs through* Connection.Start/.StartOk, Connection.Tune/.TuneOk, and then* calls Connection.Open and waits for the OpenOk. Sets heart-beat* and frame max values after tuning has taken place.* @throws IOException if an error is encountered* either before, or during, protocol negotiation;* sub-classes {@link ProtocolVersionMismatchException} and* {@link PossibleAuthenticationFailureException} will be thrown in the* corresponding circumstances. {@link AuthenticationFailureException}* will be thrown if the broker closes the connection with ACCESS_REFUSED.* If an exception is thrown, connection resources allocated can all be* garbage collected when the connection object is no longer referenced.*/public void start()throws IOException, TimeoutException {initializeConsumerWorkService();initializeHeartbeatSender();this._running = true;// Make sure that the first thing we do is to send the header,// which should cause any socket errors to show up for us, rather// than risking them pop out in the MainLoopAMQChannel.SimpleBlockingRpcContinuation connStartBlocker =new AMQChannel.SimpleBlockingRpcContinuation();// We enqueue an RPC continuation here without sending an RPC// request, since the protocol specifies that after sending// the version negotiation header, the client (connection// initiator) is to wait for a connection.start method to// arrive._channel0.enqueueRpc(connStartBlocker);try {// The following two lines are akin to AMQChannel's// transmit() method for this pseudo-RPC._frameHandler.setTimeout(handshakeTimeout);_frameHandler.sendHeader();} catch (IOException ioe) {_frameHandler.close();throw ioe;}this._frameHandler.initialize(this);AMQP.Connection.Start connStart;AMQP.Connection.Tune connTune = null;try {connStart =(AMQP.Connection.Start) connStartBlocker.getReply(handshakeTimeout/2).getMethod();_serverProperties = Collections.unmodifiableMap(connStart.getServerProperties());Version serverVersion =new Version(connStart.getVersionMajor(),connStart.getVersionMinor());if (!Version.checkVersion(clientVersion, serverVersion)) {throw new ProtocolVersionMismatchException(clientVersion,serverVersion);}String[] mechanisms = connStart.getMechanisms().toString().split(" ");SaslMechanism sm = this.saslConfig.getSaslMechanism(mechanisms);if (sm == null) {throw new IOException("No compatible authentication mechanism found - " +"server offered [" + connStart.getMechanisms() + "]");}String username = credentialsProvider.getUsername();String password = credentialsProvider.getPassword();LongString challenge = null;LongString response = sm.handleChallenge(null, username, password);do {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, handshakeTimeout/2).getMethod();if (serverResponse instanceof AMQP.Connection.Tune) {connTune = (AMQP.Connection.Tune) serverResponse;} else {challenge = ((AMQP.Connection.Secure) serverResponse).getChallenge();response = sm.handleChallenge(challenge, username, password);}} catch (ShutdownSignalException e) {Method shutdownMethod = e.getReason();if (shutdownMethod instanceof AMQP.Connection.Close) {AMQP.Connection.Close shutdownClose = (AMQP.Connection.Close) shutdownMethod;if (shutdownClose.getReplyCode() == AMQP.ACCESS_REFUSED) {throw new AuthenticationFailureException(shutdownClose.getReplyText());}}throw new PossibleAuthenticationFailureException(e);}} while (connTune == null);} catch (TimeoutException te) {_frameHandler.close();throw te;} catch (ShutdownSignalException sse) {_frameHandler.close();throw AMQChannel.wrap(sse);} catch(IOException ioe) {_frameHandler.close();throw ioe;}try {int channelMax =negotiateChannelMax(this.requestedChannelMax,connTune.getChannelMax());_channelManager = instantiateChannelManager(channelMax, threadFactory);int frameMax =negotiatedMaxValue(this.requestedFrameMax,connTune.getFrameMax());this._frameMax = frameMax;int heartbeat =negotiatedMaxValue(this.requestedHeartbeat,connTune.getHeartbeat());setHeartbeat(heartbeat);_channel0.transmit(new AMQP.Connection.TuneOk.Builder().channelMax(channelMax).frameMax(frameMax).heartbeat(heartbeat).build());_channel0.exnWrappingRpc(new AMQP.Connection.Open.Builder().virtualHost(_virtualHost).build());} catch (IOException ioe) {_heartbeatSender.shutdown();_frameHandler.close();throw ioe;} catch (ShutdownSignalException sse) {_heartbeatSender.shutdown();_frameHandler.close();throw AMQChannel.wrap(sse);}// We can now respond to errors having finished tailoring the connectionthis._inConnectionNegotiation = false;}

上面这段代码比较长，也是最核心的启动连接代码了，其实他的注释已经说的很清楚了，我们来看下注释：

* Start up the connection, including the MainLoop thread.
启动连接，包括MainLoop thread  重点来了-后续我们的消费消息，就主要靠这哥们了。
* Sends the protocol
发送协议
* version negotiation header, and runs through
协议头
* Connection.Start/.StartOk, Connection.Tune/.TuneOk, and then
这都是建立连接的相关协议内容了
* calls Connection.Open and waits for the OpenOk. 
有来有回
* Sets heart-beat 
心跳
* and frame max values after tuning has taken place.

所以，这个方法主要是发送创建连接的各种协议，双方经过沟通建立连接的过程。当然，最最最重要的一点，创建了MainLoop

创建连接的内容，我们不在过多关注了，我们主要看下创建的MainLoop，

找到：

this._frameHandler.initialize(this);

跟进去：一直跟到 AMQConnection.startMainLoop

    public void startMainLoop() {MainLoop loop = new MainLoop();final String name = "AMQP Connection " + getHostAddress() + ":" + getPort();mainLoopThread = Environment.newThread(threadFactory, loop, name);mainLoopThread.start();}

很明显MainLoop 是一个线程，通过 ThreadFactory new出来，并启动了。来看下这个线程是做什么的：

private class MainLoop implements Runnable {/*** Channel reader thread main loop. Reads a frame, and if it is* not a heartbeat frame, dispatches it to the channel it refers to.* Continues running until the "running" flag is set false by* shutdown().*/@Overridepublic void run() {boolean shouldDoFinalShutdown = true;try {while (_running) {Frame frame = _frameHandler.readFrame();readFrame(frame);}} catch (Throwable ex) {if (ex instanceof InterruptedException) {// loop has been interrupted during shutdown,// no need to do it againshouldDoFinalShutdown = false;} else {handleFailure(ex);}} finally {if (shouldDoFinalShutdown) {doFinalShutdown();}}}}

读注释即可：
* Channel reader thread main loop. Reads a frame, 
读数据帧* and if it is not a heartbeat frame, 
如果不是心跳帧* dispatches it to the channel it refers to.
分发到对应的处理channel中去* Continues running until the "running" flag is set false by shutdown().
一直在持续运行，直至关闭
总结：

这就很清楚了，这是一个无限循环的线程，一直在读取Broker传递给我们的信息，读到对应的非心跳的内容，转交到对应的处理类进行处理。

到这里我们是不是有些思路了，其实消费者的消费就是在这里监听到，并处理的。

截止到此，创建连接的内容完毕，并引出了MainLoop的内容。

5、消费者消费

好，我们继续看下这个死循环，是如何读消息的：

Frame frame = _frameHandler.readFrame();

这个是读到的消息，并包装成了Frame 对象，我们不再看这部分内容了

继续：跟进

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

上面的内容是真正的去解析处理读到的Frame 的内容了，我们看 channel.handleFrame(frame);

即可，继续追踪：

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

继续追踪处理：

command.handleFrame(frame)

这个实际上是解析消息的具体内容，然后设置到对应的对象中的属性中去了。pass掉了

继续：

handleCompleteInboundCommand(command);

追踪至 AMQChannel中的

handleCompleteInboundCommand- >

processAsync(command)

一至到ChannelN 中的 processAsync

/*** Protected API - Filters the inbound command stream, processing* Basic.Deliver, Basic.Return and Channel.Close specially.  If* we're in quiescing mode, all inbound commands are ignored,* except for Channel.Close and Channel.CloseOk.*/@Override public boolean processAsync(Command command) throws IOException{// If we are isOpen(), then we process commands normally.//// If we are not, however, then we are in a quiescing, or// shutting-down state as the result of an application// decision to close this channel, and we are to discard all// incoming commands except for a close and close-ok.Method method = command.getMethod();// we deal with channel.close in the same way, regardlessif (method instanceof Channel.Close) {asyncShutdown(command);return true;}if (isOpen()) {// We're in normal running mode.if (method instanceof Basic.Deliver) {processDelivery(command, (Basic.Deliver) method);return true;} else if (method instanceof Basic.Return) {callReturnListeners(command, (Basic.Return) method);return true;} else if (method instanceof Channel.Flow) {Channel.Flow channelFlow = (Channel.Flow) method;synchronized (_channelMutex) {_blockContent = !channelFlow.getActive();transmit(new Channel.FlowOk(!_blockContent));_channelMutex.notifyAll();}return true;} else if (method instanceof Basic.Ack) {Basic.Ack ack = (Basic.Ack) method;callConfirmListeners(command, ack);handleAckNack(ack.getDeliveryTag(), ack.getMultiple(), false);return true;} else if (method instanceof Basic.Nack) {Basic.Nack nack = (Basic.Nack) method;callConfirmListeners(command, nack);handleAckNack(nack.getDeliveryTag(), nack.getMultiple(), true);return true;} else if (method instanceof Basic.RecoverOk) {for (Map.Entry entry : Utility.copy(_consumers).entrySet()) {this.dispatcher.handleRecoverOk(entry.getValue(), entry.getKey());}// Unlike all the other cases we still want this RecoverOk to// be handled by whichever RPC continuation invoked Recover,// so return falsereturn false;} else if (method instanceof Basic.Cancel) {Basic.Cancel m = (Basic.Cancel)method;String consumerTag = m.getConsumerTag();Consumer callback = _consumers.remove(consumerTag);if (callback == null) {callback = defaultConsumer;}if (callback != null) {try {this.dispatcher.handleCancel(callback, consumerTag);} catch (WorkPoolFullException e) {// couldn't enqueue in work pool, propagatingthrow e;} catch (Throwable ex) {getConnection().getExceptionHandler().handleConsumerException(this,ex,callback,consumerTag,"handleCancel");}}return true;} else {return false;}} else {// We're in quiescing mode == !isOpen()if (method instanceof Channel.CloseOk) {// We're quiescing, and we see a channel.close-ok:// this is our signal to leave quiescing mode and// finally shut down for good. Let it be handled as an// RPC reply one final time by returning false.return false;} else {// We're quiescing, and this inbound command should be// discarded as per spec. "Consume" it by returning// true.return true;}}}

开始逐步烧脑了，这一段debug的时候，记得开启Thread模式哦。

首先看方法名吧：processAsync 明显的一个异步处理，究竟处理啥？看注释：
* Protected API - Filters the inbound command stream, processing
* Basic.Deliver, Basic.Return and Channel.Close specially.  If
* we're in quiescing mode, all inbound commands are ignored,
* except for Channel.Close and Channel.CloseOk.
处理 Basic.Deliver, Basic.Return and Channel.Close ，我们最关注的是 Deliver 投递对吧，

这就是Broker把消息投递给我们呢方法。

所以，我们找到了我们的关注点：

  if (method instanceof Basic.Deliver) {processDelivery(command, (Basic.Deliver) method);return true;

继续哦：

 protected void processDelivery(Command command, Basic.Deliver method) {Basic.Deliver m = method;Consumer callback = _consumers.get(m.getConsumerTag());if (callback == null) {if (defaultConsumer == null) {// No handler set. We should blow up as this message// needs acking, just dropping it is not enough. See bug// 22587 for discussion.throw new IllegalStateException("Unsolicited delivery -" +" see Channel.setDefaultConsumer to handle this" +" case.");}else {callback = defaultConsumer;}}Envelope envelope = new Envelope(m.getDeliveryTag(),m.getRedelivered(),m.getExchange(),m.getRoutingKey());try {// call metricsCollector before the dispatching (which is async anyway)// this way, the message is inside the stats before it is handled// in case a manual ack in the callback, the stats will be able to record the ackmetricsCollector.consumedMessage(this, m.getDeliveryTag(), m.getConsumerTag());this.dispatcher.handleDelivery(callback,m.getConsumerTag(),envelope,(BasicProperties) command.getContentHeader(),command.getContentBody());} catch (WorkPoolFullException e) {// couldn't enqueue in work pool, propagatingthrow e;} catch (Throwable ex) {getConnection().getExceptionHandler().handleConsumerException(this,ex,callback,m.getConsumerTag(),"handleDelivery");}}

哦哦，我们看到了什么 Consumer callback ，消费者回调，That's whant we want.
this.dispatcher.handleDelivery(callback,m.getConsumerTag(),envelope,(BasicProperties) command.getContentHeader(),command.getContentBody());

继续哦

进入到了ConsumerDispatcher

handleDelivery

public void handleDelivery(final Consumer delegate,final String consumerTag,final Envelope envelope,final AMQP.BasicProperties properties,final byte[] body) throws IOException {executeUnlessShuttingDown(new Runnable() {@Overridepublic void run() {try {delegate.handleDelivery(consumerTag,envelope,properties,body);} catch (Throwable ex) {connection.getExceptionHandler().handleConsumerException(channel,ex,delegate,consumerTag,"handleDelivery");}}});}

哇，handleDelivery 有没有很熟悉，我们的业务代码监听不就是实现的她吗? 真的是她吗？

是她是她就是她。。。

再来波回忆杀：

DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("小明洗澡用水: " + new String(body, "UTF-8"));}};

好了，到此我们终于找到了，是死循环读来的消息，调用回调Consumer，调用到了子类的实现的handleDelivery 方法，真正的去实现消息的消费。

不过还没完，她到底是怎么触发的呢？

 private void executeUnlessShuttingDown(Runnable r) {if (!this.shuttingDown) execute(r);}

    private void execute(Runnable r) {checkShutdown();this.workService.addWork(this.channel, r);}

继续来到

ConsumerWorkService.addWork

    public void addWork(Channel channel, Runnable runnable) {if (this.workPool.addWorkItem(channel, runnable)) {this.executor.execute(new WorkPoolRunnable());}}

继续：WorkPool

addWorkItem

  /*** Add (enqueue) an item for a specific client.* No change and returns false if client not registered.* If dormant, the client will be marked ready.* @param key the client to add to the work item to* @param item the work item to add to the client queue* @return true if and only if the client is marked ready* — as a result of this work item*/public boolean addWorkItem(K key, W item) {VariableLinkedBlockingQueue queue;synchronized (this) {queue = this.pool.get(key);}// The put operation may block. We need to make sure we are not holding the lock while that happens.if (queue != null) {enqueueingCallback.accept(queue, item);synchronized (this) {if (isDormant(key)) {dormantToReady(key);return true;}}}return false;}

这稍微有点绕了，首先我们要从 Map 缓存pool 取出一个 VariableLinkedBlockingQueue

根据啥取呢，根据的是Channel，所以每个Channel是相独立的，Blocking Queue后续的操作也是阻塞的。

来到了
enqueueingCallback.accept(queue, item);
这是个啥鬼？这又一个回调，使用了@FunctionalInterface

真实的方法在初始WorkPool 的时候

 public WorkPool(final int queueingTimeout) {if (queueingTimeout > 0) {this.enqueueingCallback = (queue, item) -> {try {boolean offered = queue.offer(item, queueingTimeout, TimeUnit.MILLISECONDS);if (!offered) {throw new WorkPoolFullException("Could not enqueue in work pool after " + queueingTimeout + " ms.");}} catch (InterruptedException e) {Thread.currentThread();}};} else {this.enqueueingCallback = (queue, item) -> {try {queue.put(item);} catch (InterruptedException e) {Thread.currentThread().interrupt();}};}}

我们看后面的else内容即可：
queue.put(item);
What ?竟然把内容放到了一个本地的BlockingQueue 中去了，放的啥内容呢？

就是我们前面的那个线程对象
Runnable runnable
倒腾一下嘛，就是那个
delegate.handleDelivery(consumerTag,envelope,properties,body);
有点意思了吧，把整个需要消费的内容扔进了队列里，这时候程序就可以返回给MainLoop了，他有可以继续抓包了。

但我们还没结束，对吧，继续咯

代码倒回来：

  if (this.workPool.addWorkItem(channel, runnable)) {this.executor.execute(new WorkPoolRunnable());}

这次要看

executor.execute了

首先我们看下这个 ExecutorService executor 这个没特殊指定的话，我们再初始Connection的时候就会初始化这个 ConsumerWorkService，就把 executor 初始化了，一个固定的线程池：

public ConsumerWorkService(ExecutorService executor, ThreadFactory threadFactory, int queueingTimeout, int shutdownTimeout) {this.privateExecutor = (executor == null);this.executor = (executor == null) ? Executors.newFixedThreadPool(DEFAULT_NUM_THREADS, threadFactory): executor;this.workPool = new WorkPool<>(queueingTimeout);this.shutdownTimeout = shutdownTimeout;}

几个线程呢？
private static final int DEFAULT_NUM_THREADS = Runtime.getRuntime().availableProcessors() * 2;
当前计算机的核数 * 2 ，八核的就是初始化 16个线程。

这16个线程是跟随Connection的，所以，每个Connection就只有这16个线程在处理呗。

继续咯

this.executor.execute(new WorkPoolRunnable());

又要搞个线程，

 private final class WorkPoolRunnable implements Runnable {@Overridepublic void run() {int size = MAX_RUNNABLE_BLOCK_SIZE;List block = new ArrayList(size);try {Channel key = ConsumerWorkService.this.workPool.nextWorkBlock(block, size);if (key == null) return; // nothing ready to runtry {for (Runnable runnable : block) {runnable.run();}} finally {if (ConsumerWorkService.this.workPool.finishWorkBlock(key)) {ConsumerWorkService.this.executor.execute(new WorkPoolRunnable());}}} catch (RuntimeException e) {Thread.currentThread().interrupt();}}}

终于等到你，这就是我们核心中的核心了，触发消费也就靠这了。

这个线程被线程池搞起后，做啥了呢？

1.声明一个 16个大小的 ArrayList

2. 取出我们的BlockingQueue，再接着呢，从Queue中取出16个Runnable对象【真正的消费逻辑】，放到ArrayList 中

3. 循环16个 Runable对象，直接调用其run 方法，这时候自然就调到了我们的handleDelivery- 业务方法愉快的去消费了。

4. 最后呢，还要看我们这队列中还没有待处理的数据了，如果还要有的话，通过线程池再起线程继续执行 WorkPoolRunnable 的run 方法，也就是本方法，

如果队列中一直有消息，而且还一直有消息进来，那线程池就会一直在启线程处理，直到16个线程都启动满负载运转，这时候就会存在本地BlockingQueue的堆积了。

补充下消费的抓包情况：