devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

[RabbitMQ]Java客户端：源码概览_rabbitmq java客户端

作者：笔触狂放9 | 2024-04-08 09:15:15

赞

踩

rabbitmq java客户端

本文简要介绍RabbitMQ提供的Java客户端中最基本的功能性接口/类及相关源码。

Mavan依赖：

  1. <dependency>
  2.  
  3.     <groupId>com.rabbitmq</groupId>
  4.  
  5.     <artifactId>amqp-client</artifactId>
  6.  
  7.     <version>5.13.1</version>
  8.  
  9. </dependency>

0 AMQP#

com.rabbitmq.client.AMQP接口将AMQP（Advanced Message Queue Protocol，高级消息队列协议）中的方法和消息属性封装成Java对象，便于以面向对象的思维模式进行编程。

该接口类图简要如下：

AMQP接口中包含许多内部类，大体可以分成三类：

0.1 协议信息#

PROTOCOL内部类，保存了AMQP的协议版本等信息。

  1. public static class PROTOCOL {
  2.  
  3.     public static final int MAJOR = 0;
  4.  
  5.     public static final int MINOR = 9;
  6.  
  7.     public static final int REVISION = 1;
  8.  
  9.     public static final int PORT = 5672;
  10.  
  11. }

0.2 方法#

包括Connection、Channel、Access、Exchange、Queue、Basic、Tx和Confirm内部类，分别封装了向Broker发送的不同方法的基本数据格式和内容。

它们都实现了com.rabbitmq.client.impl.Method抽象类（后续会介绍），在发送请求时，通过Method抽象类的toFrame()方法可以转换成Frame（帧），然后com.rabbitmq.client.impl.AMQConnection将其以二进制数据的方式通过TCP协议发送给Broker。

它们都提供了各自的Builder，便于实例化方法对象（建造者模式）。例如，最常用的Publish方法类图简要如下：

通过如下代码实例化出Publish对象：

AMQP.Basic.Publish publish = new AMQP.Basic.Publish.Builder().exchange("myExchange").routingKey("myRoutingKey").mandatory(true).build();

在发送给Broker前可以通过如下代码将Publish对象转成帧：

  1. Method method = (Method) publish;
  2.  
  3. Frame frame = method.toFrame(1);

com.rabbitmq.client.impl.AMQConnection对象管理着与Broker的连接，它通过如下代码将方法发送给Broker：

  1. AMQConnection connection = channel.getConnection();
  2.  
  3. connection.writeFrame(frame);

0.3 消息属性#

BasicProperties内部类，封装了消息的基本属性。

它也提供了Builder，我们在发送消息时可以使用BasicProperties实例携带消息头信息，类图如下：

通过如下代码实例化出BasicProperties对象，并发送消息：

  1. Map<String, Object> headers = new HashMap<>();
  2.  
  3. headers.put("color", "blue");
  4.  
  5. AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().headers(headers).expiration("10000").build();
  6.  
  7. channel.basicPublish("myExchange", "myRoutingKey", true, properties, "hello".getBytes());

BasicProperties对象在发送前最终会被转换成com.rabbitmq.client.impl.AMQContentHeader对象，代表AMQ消息内容的头。

AMQContentHeader的toFrame()方法也可以将其转换成Frame（帧），然后com.rabbitmq.client.impl.AMQConnection将其以二进制数据的方式通过TCP协议发送给Broker。

  1. AMQConnection connection = channel.getConnection();
  2.  
  3. Frame headerFrame = contentHeader.toFrame(channelNumber, body.length);
  4.  
  5. connection.writeFrame(headerFrame);

1 ConnectionFactory#

com.rabbitmq.client.ConnectionFactory类是用来创建与RabbitMQ服务器连接（com.rabbitmq.client.Connection）的工厂类。简要类图如下：

ConnectionFactory内部封装了许多属性，用来设置与Connection或Socket相关的连接信息。

它还提供了一套默认配置：

  1. public static final String DEFAULT_VHOST = "/";
  2.  
  3. public static final String DEFAULT_HOST = "localhost";
  4.  
  5. public static final int    DEFAULT_AMQP_PORT = AMQP.PROTOCOL.PORT; // 5672
  6.  
  7. public static final int    DEFAULT_AMQP_OVER_SSL_PORT = 5671;
  8.  
  9. public static final String DEFAULT_PASS = "guest";	// CredentialsProvider.username
  10.  
  11. public static final String DEFAULT_USER = "guest";	// CredentialsProvider.password
  12.  
  13. private boolean nio = false;
  14.  
  15. private boolean automaticRecovery               = true;

ConnectionFactory的基本使用如下：

  1. ConnectionFactory connectionFactory = new ConnectionFactory();
  2.  
  3. Connection connection = connectionFactory.newConnection();//返回RecoveryAwareAMQConnection或AMQConnection对象

底层会创建出java.net.Socket或java.nio.channels.SocketChannel，代表与RabbitMQ服务器的TCP连接：

  1. Socket socket = SocketFactory.getDefault().createSocket();
  2.  
  3. SocketChannel channel = SocketChannel.open();

Socket或SocketChannel会被封装到com.rabbitmq.client.impl.FrameHandler中：

  1. // nio==false，使用Socket（默认值）
  2.  
  3. SocketFrameHandler frameHandler = new SocketFrameHandler(sock, this.shutdownExecutor);
  4.  
  5. // nio==true，使用SocketChannel
  6.  
  7. SocketChannelFrameHandlerState state = new SocketChannelFrameHandlerState(channel, nioLoopContext, nioParams, sslEngine);
  8.  
  9. SocketChannelFrameHandler frameHandler = new SocketChannelFrameHandler(state);

FrameHandler中提供了readFrame()和writeFrame()，分别可以从Socket/SocketChannel中读取或写入数据。

FrameHandler又会被封装到com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnection或com.rabbitmq.client.impl.AMQConnection中：

  1. // automaticRecovery==true，默认值
  2.  
  3. FrameHandler frameHandler = factory.create(addr, connectionName());
  4.  
  5. RecoveryAwareAMQConnection conn = createConnection(params, frameHandler, metricsCollector);
  6.  
  7. // automaticRecovery==false
  8.  
  9. FrameHandler handler = fhFactory.create(addr, clientProvidedName);
  10.  
  11. AMQConnection conn = createConnection(params, handler, metricsCollector);

因此，我们可以使用返回的Connection对象与RabbitMQ服务器进行交互。

2 Connection#

com.rabbitmq.client.Connection接口代表与RabbitMQ服务器的TCP连接，类图简要如下：

Connection主要提供了createChannel()和openChannel()方法，用来创建Channel。后者提供了几乎所有与RabbitMQ进行交互的方法，是项目中使用频率最高的一个接口。

Connection的基本使用如下：

Channe channel = connection.createChannel();

Connection的实现类主要包括以下几种，分别代表不同类型的连接：

  • AMQConnection类：代表最基本的与RabbitMQ服务器的连接。内部持有FrameHandler等成员变量，用来与服务器交互。
  • RecoveryAwareAMQConnection接口：代表自动重连的连接，内部没有方法，类似与标志性接口。
  • AutorecoveringConnection类：自动重连Connection的实现类，在非正常断开情况下会自动重连，例如I/O异常。它持有RecoveryAwareAMQConnection对象作为代理，从而间接可以使用FrameHandler对象与服务器进行交互。重连时，内部组件也会按如下顺序自动重连：
    • Exchanges
    • Queues
    • Bindings (both queue and exchange-to-exchange)
    • Consumers
  • RecoveryAwareAMQConnection：它是对AMQConnection的修改，主要用作AutorecoveringConnection的成员变量。它与AMQConnection主要区别在于它内部使用com.rabbitmq.client.impl.recovery.RecoveryAwareChannelN作为Channel。

在项目中使用的实现类主要为AMQConnection和AutorecoveringConnection（根据ConnectionFactory的automaticRecovery成员变量进行选择）。

AMQConnection.createChannel()方法会使用ChannelManager创建出ChannelN类型的通道：

  1. ChannelManager cm = _channelManager;
  2.  
  3. Channel channel = cm.createChannel(this);
  4.  
  5. // 底层：
  6.  
  7. new ChannelN(connection, channelNumber, workService, this.metricsCollector);

而AutorecoveringConnection.createChannel()方法会使用RecoveryAwareAMQConnection创建出RecoveryAwareChannelN类型的通道，并使用AutorecoveringChannel包装：

  1. RecoveryAwareChannelN ch = (RecoveryAwareChannelN) delegate.createChannel();
  2.  
  3. final AutorecoveringChannel channel = new AutorecoveringChannel(this, delegateChannel);

ChannelManager是AMQConnection中一个十分重要的成员变量，它管理着AMQConnection对象所属的所有Channel对象（key为通道编号，取值范围为1~_channelMax）：

private final Map<Integer, ChannelN> _channelMap = new HashMap<Integer, ChannelN>();

AMQConnection中有一个十分重要的方法writeFrame()，可以将数据发送给RabbitMQ服务器：

  1. public void writeFrame(Frame f) throws IOException {
  2.  
  3.     _frameHandler.writeFrame(f);
  4.  
  5.     _heartbeatSender.signalActivity();
  6.  
  7. }
  8.  
  9.  
  10.  
  11. // SocketFrameHandler
  12.  
  13. public void writeFrame(Frame frame) throws IOException {
  14.  
  15.     synchronized (_outputStream) {
  16.  
  17.         frame.writeTo(_outputStream);
  18.  
  19.     }
  20.  
  21. }
  22.  
  23. public void writeTo(DataOutputStream os) throws IOException {
  24.  
  25.     os.writeByte(type);
  26.  
  27.     os.writeShort(channel);
  28.  
  29.     if (accumulator != null) {
  30.  
  31.         os.writeInt(accumulator.size());
  32.  
  33.         accumulator.writeTo(os);
  34.  
  35.     } else {
  36.  
  37.         os.writeInt(payload.length);
  38.  
  39.         os.write(payload);
  40.  
  41.     }
  42.  
  43.     os.write(AMQP.FRAME_END);
  44.  
  45. }

AMQConnection中有一个十分重要的方法startMainLoop()，可以创建新线程监听RabbitMQ服务器发送来的消息：

  1. public void startMainLoop() {
  2.  
  3.     MainLoop loop = new MainLoop();
  4.  
  5.     final String name = "AMQP Connection " + getHostAddress() + ":" + getPort();
  6.  
  7.     mainLoopThread = Environment.newThread(threadFactory, loop, name);
  8.  
  9.     mainLoopThread.start();
  10.  
  11. }

其核心在于MainLoop内部类，核心步骤如下：

  1. 调用Frame frame = _frameHandler.readFrame()读取RabbitMQ服务器发送来的消息。
  2. 调用readFrame(frame)处理消息。
  1. private class MainLoop implements Runnable {
  2.  
  3.     @Override
  5.     public void run() {
  7.         boolean shouldDoFinalShutdown = true;
  9.         try {
  11.             while (_running) {
  13.                 Frame frame = _frameHandler.readFrame();
  15.                 readFrame(frame);
  17.             }
  19.         } catch (Throwable ex) {
  21.             if (ex instanceof InterruptedException) {
  23.                 shouldDoFinalShutdown = false;
  25.             } else {
  27.                 handleFailure(ex);
  29.             }
  31.         } finally {
  33.             if (shouldDoFinalShutdown) {
  35.                 doFinalShutdown();
  37.             }
  39.         }
  41.     }
  43. }

readFrame(frame)方法会从ChannelManager成员变量中获取该消息对应的通道，然后调用channel.handleFrame(frame)方法进行业务处理（最终会调用channel.processAsync(Command command)方法）：

  1. private void readFrame(Frame frame) throws IOException {
  2.  
  3.     if (frame != null) {
  5.         _missedHeartbeats = 0;
  7.         if (frame.type == AMQP.FRAME_HEARTBEAT) {
  9.         } else {
  11.             if (frame.channel == 0) {
  13.                 _channel0.handleFrame(frame);
  15.             } else {
  17.                 if (isOpen()) {
  19.                     ChannelManager cm = _channelManager;
  21.                     if (cm != null) {
  23.                         ChannelN channel;
  25.                         try {
  27.                             channel = cm.getChannel(frame.channel);
  29.                         } catch(UnknownChannelException e) {
  31.                             LOGGER.info("Received a frame on an unknown channel, ignoring it");
  33.                             return;
  35.                         }
  37.                         channel.handleFrame(frame);	// 业务处理
  39.                     }
  41.                 }
  43.             }
  45.         }
  47.     } else {
  49.         handleSocketTimeout();
  51.     }
  53. }

3 Channel#

com.rabbitmq.client.Channel中封装了与RabbitMQ服务器交互的API，简要类图如下：

Channel的基本使用方式如下：

  1. // 声明交换机
  2.  
  3. channel.exchangeDeclare("myExchange", BuiltinExchangeType.DIRECT);
  4.  
  5. // 声明队列
  6.  
  7. channel.queueDeclare("myQueue", true, false, false, null);
  8.  
  9. // 声明绑定
  10.  
  11. channel.exchangeBind("myQueue", "myExchange", "myRoutingKey");
  12.  
  13. // 发送消息
  14.  
  15. channel.basicPublish("myExchange", "myRoutingKey", true, null, "hello".getBytes());
  16.  
  17. // 订阅消息
  18.  
  19. channel.basicConsume("myQueue", new DefaultConsumer(channel) {
  20.  
  21.             @Override
  22.  
  23.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  24.  
  25.                 System.out.println(new String(body));
  26.  
  27.             }
  28.  
  29.         });
  30.  
  31. // 拉取消息
  32.  
  33. channel.basicGet("myQueue", true);

Channel是实现类包括以下几种：

  • ChannelN：AMQP协议功能API的主要实现类。
  • RecoveryAwareChannelN：重写了basicAck()、basicReject()和basicNack()方法，对ChannelN功能进行扩展，实时跟踪delivery tag，对最新的tag进行响应。
  • AutorecoveringChannel：在connection重连时会自动恢复的通道，内部通过持有RecoveryAwareChannelN代理对象来实现具体操作。

3.1 ChannelN#

com.rabbitmq.client.impl.ChannelN是对AMQP协议功能性API的主要实现类，它除了实现Channel中定义的AMQP协议功能性API，还继承了AMQChannel抽象类，通过其_connection成员变量可以在底层调用到Socket或SocketChannel向RabbitMQ服务器进行读写操作。

除此之外，为了实现AMQP协议的特定功能，如消息确认机制。ChannelN内部封装了如下成员变量：

  • _consumers：消息消费者，以consumerTag作为key，用于监听消息。
  • returnListeners：监听RabbitMQ服务器找不到对应交换机时的返回消息（basicPublish方法发送消息时设置mandatory或immediate）。
  • confirmListeners：监听RabbitMQ服务器的确认消息（ack或nack）。
  • defaultConsumer：默认的消息消费者。
  • dispatcher：启动线程执行_consumers中的任务。

ChannelN中监听消息的核心源码如下：

  1. public boolean processAsync(Command command) throws IOException {
  2.  
  3.     Method method = command.getMethod();
  5.     if (method instanceof Channel.Close) {
  7.         asyncShutdown(command);
  9.         return true;
  11.     }
  13.     if (isOpen()) {
  15.         // 根据不同方法类型调用对应的处理方法
  17.         if (method instanceof Basic.Deliver) {
  19.             processDelivery(command, (Basic.Deliver) method);
  21.             return true;
  23.         } else if (method instanceof Basic.Return) {
  25.             callReturnListeners(command, (Basic.Return) method);
  27.             return true;
  29.         } else if (method instanceof Channel.Flow) {
  31.             Channel.Flow channelFlow = (Channel.Flow) method;
  33.             synchronized (_channelMutex) {
  35.                 _blockContent = !channelFlow.getActive();
  37.                 transmit(new Channel.FlowOk(!_blockContent));
  39.                 _channelMutex.notifyAll();
  41.             }
  43.             return true;
  45.         } else if (method instanceof Basic.Ack) {
  47.             Basic.Ack ack = (Basic.Ack) method;
  49.             callConfirmListeners(command, ack);
  51.             handleAckNack(ack.getDeliveryTag(), ack.getMultiple(), false);
  53.             return true;
  55.         } else if (method instanceof Basic.Nack) {
  57.             Basic.Nack nack = (Basic.Nack) method;
  59.             callConfirmListeners(command, nack);
  61.             handleAckNack(nack.getDeliveryTag(), nack.getMultiple(), true);
  63.             return true;
  65.         } else if (method instanceof Basic.RecoverOk) {
  67.             for (Map.Entry<String, Consumer> entry : Utility.copy(_consumers).entrySet()) {
  69.                 this.dispatcher.handleRecoverOk(entry.getValue(), entry.getKey());
  71.             }
  73.             return false;
  75.         } else if (method instanceof Basic.Cancel) {
  77.             Basic.Cancel m = (Basic.Cancel)method;
  79.             String consumerTag = m.getConsumerTag();
  81.             Consumer callback = _consumers.remove(consumerTag);
  83.             if (callback == null) {
  85.                 callback = defaultConsumer;
  87.             }
  89.             if (callback != null) {
  91.                 try {
  93.                     this.dispatcher.handleCancel(callback, consumerTag);
  95.                 } catch (WorkPoolFullException e) {
  97.                     throw e;
  99.                 } catch (Throwable ex) {
  101.                     getConnection().getExceptionHandler().handleConsumerException(this,
  103.                                                                                   ex,
  105.                                                                                   callback,
  107.                                                                                   consumerTag,
  109.                                                                                   "handleCancel");
  111.                 }
  113.             } else {
  115.                 LOGGER.warn("Could not cancel consumer with unknown tag {}", consumerTag);
  117.             }
  119.             return true;
  121.         } else {
  123.             return false;
  125.         }
  127.     } else {
  129.         if (method instanceof Channel.CloseOk) {
  131.             return false;
  133.         } else {
  135.             return true;
  137.         }
  139.     }
  141. }

可见，该方法类似于SpringMVC中的DispatcherServlet，它会根据监听到Command对象的方法类型进行分发处理。接下来介绍的各成员变量方法调用的入口都在这个方法中。

3.1.1 ConsumerDispatcher#

com.rabbitmq.client.impl.ConsumerDispatcher的作用是从线程池中获取空闲线程处理消息。它的主要作用是开启线程，而实际处理消息的业务逻辑在具体Consumer代理对象中处理。

例如，在处理生产者发布的消息时，ConsumerDispatcher会进行如下处理：

  1. public void handleDelivery(final Consumer delegate,
  2.  
  3.                            final String consumerTag,
  5.                            final Envelope envelope,
  7.                            final AMQP.BasicProperties properties,
  9.                            final byte[] body) throws IOException {
  11.     executeUnlessShuttingDown(
  13.     new Runnable() {
  15.         @Override
  17.         public void run() {
  19.             try {
  21.                 delegate.handleDelivery(consumerTag,
  23.                         envelope,
  25.                         properties,
  27.                         body);
  29.             } catch (Throwable ex) {
  31.                 connection.getExceptionHandler().handleConsumerException(
  33.                         channel,
  35.                         ex,
  37.                         delegate,
  39.                         consumerTag,
  41.                         "handleDelivery");
  43.             }
  45.         }
  47.     });
  49. }

3.1.2 Consumer#

com.rabbitmq.client.Consumer接口中定义了不同消息的处理方法，实例对象则表示消息消费者。

com.rabbitmq.client.DefaultConsumer是默认实现类，它实现了接口中的所有方法（空方法）。我们可以采取匿名内部类的方式，实现具体某个需要的方法，而不是实现所有方法。

我们可以使用如下代码添加消费者：

  1. channel.basicConsume("myQueue", new DefaultConsumer(channel) {
  2.  
  3.     @Override
  4.  
  5.     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  6.  
  7.         System.out.println(new String(body));
  8.  
  9.     }
  10.  
  11. });

在ChannelN中调用消费者处理消息方法（handleDelivery()）的源码如下：

  1. protected void processDelivery(Command command, Basic.Deliver method) {
  2.  
  3.     Basic.Deliver m = method;
  5.     Consumer callback = _consumers.get(m.getConsumerTag());
  7.     if (callback == null) {
  9.         if (defaultConsumer == null) {
  11.             throw new IllegalStateException("Unsolicited delivery -" +
  13.                     " see Channel.setDefaultConsumer to handle this" +
  15.                     " case.");
  17.         }
  19.         else {
  21.             callback = defaultConsumer;
  23.         }
  25.     }
  27.  
  28.  
  29.     Envelope envelope = new Envelope(m.getDeliveryTag(),
  31.                                      m.getRedelivered(),
  33.                                      m.getExchange(),
  35.                                      m.getRoutingKey());
  37.     try {
  39.         metricsCollector.consumedMessage(this, m.getDeliveryTag(), m.getConsumerTag());
  41.         this.dispatcher.handleDelivery(callback,
  43.                                        m.getConsumerTag(),
  45.                                        envelope,
  47.                                        (BasicProperties) command.getContentHeader(),
  49.                                        command.getContentBody());
  51.     } catch (WorkPoolFullException e) {
  53.         throw e;
  55.     } catch (Throwable ex) {
  57.         getConnection().getExceptionHandler().handleConsumerException(this,
  59.             ex,
  61.             callback,
  63.             m.getConsumerTag(),
  65.             "handleDelivery");
  67.     }
  69. }

需要注意的是，在调用this.dispatcher.handleDelivery()之前，会首先调用Consumer callback = _consumers.get(m.getConsumerTag())根据consumerTag获取对应的消费者。因此，消费者处理消息是一对一的。

消费者其他方法的调用也可以在ChannelN.processAsync()中找到。

3.1.3 ReturnListener#

com.rabbitmq.client.ReturnListener接口中定义了监听返回消息的通用方法handleReturn()，主要用于消息发布者监听返回消息。

消息发布者通过basicPublish方法发送消息时设置mandatory或immediate，但RabbitMQ服务器找不到对应交换机时会返回消息。消息发布者通过往Channel对象中添加ReturnListener实现类，即可监听到返回消息：

  1. channel.addReturnListener(new ReturnListener() {
  2.  
  3.     @Override
  4.  
  5.     public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
  6.  
  7.         System.out.println("return message: " + new String(body));
  8.  
  9.     }
  10.  
  11. });

在ChannelN中处理返回消息的源码如下：

  1. private void callReturnListeners(Command command, Basic.Return basicReturn) {
  2.  
  3.     try {
  4.  
  5.         for (ReturnListener l : this.returnListeners) {
  6.  
  7.             l.handleReturn(basicReturn.getReplyCode(),
  8.  
  9.                 basicReturn.getReplyText(),
  10.  
  11.                 basicReturn.getExchange(),
  12.  
  13.                 basicReturn.getRoutingKey(),
  14.  
  15.                 (BasicProperties) command.getContentHeader(),
  16.  
  17.                 command.getContentBody());
  18.  
  19.         }
  20.  
  21.     } catch (Throwable ex) {
  22.  
  23.         getConnection().getExceptionHandler().handleReturnListenerException(this, ex);
  24.  
  25.     } finally {
  26.  
  27.         metricsCollector.basicPublishUnrouted(this);
  28.  
  29.     }
  30.  
  31. }

ReturnListener是针对ChannelN级别的。接收到返回消息后，所有添加到ChannelN对象的ReturnListener监听器都会被调用。

3.1.4 ConfirmListener#

com.rabbitmq.client.ConfirmListener接口中定义的监听RabbitMQ服务器确认消息（ack或nack）的回调方法，主要用于消息发布者使用。

基本使用代码如下：

  1. channel.addConfirmListener(new ConfirmListener() {
  2.  
  3.     @Override
  4.  
  5.     public void handleAck(long deliveryTag, boolean multiple) throws IOException {
  6.  
  7.         // 业务处理
  8.  
  9.     }
  10.  
  11.  
  12.  
  13.     @Override
  14.  
  15.     public void handleNack(long deliveryTag, boolean multiple) throws IOException {
  16.  
  17.         // 业务处理
  18.  
  19.     }
  20.  
  21. });

在ChannelN中处理返回消息的源码如下：

  1. private void callConfirmListeners(@SuppressWarnings("unused") Command command, Basic.Ack ack) {
  2.  
  3.     try {
  4.  
  5.         for (ConfirmListener l : this.confirmListeners) {
  6.  
  7.             l.handleAck(ack.getDeliveryTag(), ack.getMultiple());
  8.  
  9.         }
  10.  
  11.     } catch (Throwable ex) {
  12.  
  13.         getConnection().getExceptionHandler().handleConfirmListenerException(this, ex);
  14.  
  15.     } finally {
  16.  
  17.         metricsCollector.basicPublishAck(this, ack.getDeliveryTag(), ack.getMultiple());
  18.  
  19.     }
  20.  
  21. }
  22.  
  23.  
  24.  
  25. private void callConfirmListeners(@SuppressWarnings("unused") Command command, Basic.Nack nack) {
  26.  
  27.     try {
  28.  
  29.         for (ConfirmListener l : this.confirmListeners) {
  30.  
  31.             l.handleNack(nack.getDeliveryTag(), nack.getMultiple());
  32.  
  33.         }
  34.  
  35.     } catch (Throwable ex) {
  36.  
  37.         getConnection().getExceptionHandler().handleConfirmListenerException(this, ex);
  38.  
  39.     } finally {
  40.  
  41.         metricsCollector.basicPublishNack(this, nack.getDeliveryTag(), nack.getMultiple());
  42.  
  43.     }
  44.  
  45. }

ConfirmListener是针对ChannelN级别的。接收到确认消息后，所有添加到ChannelN对象的ConfirmListener监听器都会被调用。

3.1.5 basicPublish()#

前几小节讲述的都是RabbitMQ客户端监听从服务器响应的消息，本小节简要分析客户端发送消息的流程。

发送消息的基本方式如下：

channel.basicPublish("myExchange", "myRoutingKey", null, "hello".getBytes());
  1. 在ChannelN中的basicPublish()方法中执行如下代码，核心步骤如下：
    1. 将形参转换成AMQCommand对象中的CommandAssembler成员变量：exchange和routingKey→Basic.Publish方法对象（Method），properties→AMQContentHeader对象，body→List<byte[]>对象。
    2. 调用transmit(command)方法，发送命令。
  1. public void basicPublish(String exchange, String routingKey,
  2.  
  3.                          boolean mandatory, boolean immediate,
  5.                          BasicProperties props, byte[] body)
  7.     throws IOException
  9. {
  10.  
  11.     if (nextPublishSeqNo > 0) {
  13.         unconfirmedSet.add(getNextPublishSeqNo());
  15.         nextPublishSeqNo++;
  17.     }
  19.     if (props == null) {
  21.         props = MessageProperties.MINIMAL_BASIC;
  23.     }
  25.     AMQCommand command = new AMQCommand(
  27.         new Basic.Publish.Builder()
  29.             .exchange(exchange)
  31.             .routingKey(routingKey)
  33.             .mandatory(mandatory)
  35.             .immediate(immediate)
  37.             .build(), props, body);
  39.     try {
  41.         transmit(command);
  43.     } catch (IOException e) {
  45.         metricsCollector.basicPublishFailure(this, e);
  47.         throw e;
  49.     }
  51.     metricsCollector.basicPublish(this);
  53. }
  1. 在ChannelN中执行transmit()和quiescingTransmit()方法，最终会调用AMQCommand.transmit()方法：
  1. public void transmit(AMQCommand c) throws IOException {
  2.  
  3.     synchronized (_channelMutex) {
  4.  
  5.         ensureIsOpen();
  6.  
  7.         quiescingTransmit(c);
  8.  
  9.     }
  10.  
  11. }
  12.  
  13.  
  14.  
  15. public void quiescingTransmit(AMQCommand c) throws IOException {
  16.  
  17.     synchronized (_channelMutex) {
  18.  
  19.         if (c.getMethod().hasContent()) {
  20.  
  21.             while (_blockContent) {
  22.  
  23.                 try {
  24.  
  25.                     _channelMutex.wait();
  26.  
  27.                 } catch (InterruptedException ignored) {
  28.  
  29.                     Thread.currentThread().interrupt();
  30.  
  31.                 }
  32.  
  33.                 ensureIsOpen();
  34.  
  35.             }
  36.  
  37.         }
  38.  
  39.         this._trafficListener.write(c);
  40.  
  41.         c.transmit(this);
  42.  
  43.     }
  44.  
  45. }
  1. 在AMQCommand中执行transmit()方法，核心步骤如下：
    1. 获取AMQConnection对象。
    2. 分别将AMQContentHeader、Method和List<byte[]>对象转换成Frame对象。
    3. 通过AMQConnection对象发送数据。
  1. public void transmit(AMQChannel channel) throws IOException {
  2.  
  3.     int channelNumber = channel.getChannelNumber();
  4.  
  5.     AMQConnection connection = channel.getConnection();
  6.  
  7.  
  8.  
  9.     synchronized (assembler) {
  10.  
  11.         Method m = this.assembler.getMethod();
  12.  
  13.         if (m.hasContent()) {
  14.  
  15.             byte[] body = this.assembler.getContentBody();
  16.  
  17.  
  18.  
  19.             Frame headerFrame = this.assembler.getContentHeader().toFrame(channelNumber, body.length);
  20.  
  21.  
  22.  
  23.             int frameMax = connection.getFrameMax();
  24.  
  25.             boolean cappedFrameMax = frameMax > 0;
  26.  
  27.             int bodyPayloadMax = cappedFrameMax ? frameMax - EMPTY_FRAME_SIZE : body.length;
  28.  
  29.             if (cappedFrameMax && headerFrame.size() > frameMax) {
  30.  
  31.                 String msg = String.format("Content headers exceeded max frame size: %d > %d", headerFrame.size(), frameMax);
  32.  
  33.                 throw new IllegalArgumentException(msg);
  34.  
  35.             }
  36.  
  37.             connection.writeFrame(m.toFrame(channelNumber));
  38.  
  39.             connection.writeFrame(headerFrame);
  40.  
  41.             for (int offset = 0; offset < body.length; offset += bodyPayloadMax) {
  42.  
  43.                 int remaining = body.length - offset;
  44.  
  45.                 int fragmentLength = (remaining < bodyPayloadMax) ? remaining
  46.  
  47.                         : bodyPayloadMax;
  48.  
  49.                 Frame frame = Frame.fromBodyFragment(channelNumber, body,
  50.  
  51.                         offset, fragmentLength);
  52.  
  53.                 connection.writeFrame(frame);
  54.  
  55.             }
  56.  
  57.         } else {
  58.  
  59.             connection.writeFrame(m.toFrame(channelNumber));
  60.  
  61.         }
  62.  
  63.     }
  64.  
  65.  
  66.  
  67.     connection.flush();
  68.  
  69. }

3.1.6 basicGet()#

除了添加Comsumer监听器，我们还可以主动调用basicGet()向RabbitMQ服务器“拉取”消息。

basicGet()方法本质上是向RabbitMQ服务器发送一个Basic.Get请求，然后等待响应。

basicGet()方法的基本使用如下：

  1. // 自动确认模式
  2.  
  3. GetResponse message = channel.basicGet("myQueue", true);
  4.  
  5. System.out.println(new String(message.getBody()));
  6.  
  7. // 手动确认模式
  8.  
  9. GetResponse myQueue = channel.basicGet("myQueue", false);
  10.  
  11. System.out.println(new String(message.getBody()));
  12.  
  13. channel.basicAck(myQueue.getEnvelope().getDeliveryTag(), false);

  1. 在ChannelN中的basicGet()方法中执行如下代码，核心步骤如下：

    1. 将形参转换成AMQCommand对象中的CommandAssembler成员变量：queue→Basic.Get方法对象（Method），properties→AMQContentHeader对象，body→List<byte[]>对象。
    2. 调用exnWrappingRpc(command)方法，发送命令。
    3. 等待响应replyCommand，并封装成GetResponse对象返回，
  1.    public GetResponse basicGet(String queue, boolean autoAck) throws IOException{
  2.  
  3.     validateQueueNameLength(queue);
  4.  
  5.     AMQCommand replyCommand = exnWrappingRpc(new Basic.Get.Builder()
  6.  
  7.                                               .queue(queue)
  8.  
  9.                                               .noAck(autoAck)
  10.  
  11.                                              .build());
  12.  
  13.     Method method = replyCommand.getMethod();
  14.  
  15.     if (method instanceof Basic.GetOk) {
  16.  
  17.         Basic.GetOk getOk = (Basic.GetOk)method;
  18.  
  19.         Envelope envelope = new Envelope(getOk.getDeliveryTag(),
  20.  
  21.                                          getOk.getRedelivered(),
  22.  
  23.                                          getOk.getExchange(),
  24.  
  25.                                          getOk.getRoutingKey());
  26.  
  27.         BasicProperties props = (BasicProperties)replyCommand.getContentHeader();
  28.  
  29.         byte[] body = replyCommand.getContentBody();
  30.  
  31.         int messageCount = getOk.getMessageCount();
  32.  
  33.         metricsCollector.consumedMessage(this, getOk.getDeliveryTag(), autoAck);
  34.  
  35.         return new GetResponse(envelope, props, body, messageCount);
  36.  
  37.     } else if (method instanceof Basic.GetEmpty) {
  38.  
  39.         return null;
  40.  
  41.     } else {
  42.  
  43.         throw new UnexpectedMethodError(method);
  44.  
  45.     }
  46.  
  47. }
  48.  
  49. public AMQCommand exnWrappingRpc(Method m) throws IOException {
  50.  
  51.     try {
  52.  
  53.         return privateRpc(m);
  54.  
  55.     } catch (AlreadyClosedException ace) {
  56.  
  57.         throw ace;
  58.  
  59.     } catch (ShutdownSignalException ex) {
  60.  
  61.         throw wrap(ex);
  62.  
  63.     }
  64.  
  65. }

  1. 在ChannelN中的privateRpc()方法中执行如下代码，核心步骤如下：

    1. 实例化SimpleBlockingRpcContinuation对象，用于获取响应。
    2. 调用rpc(m, k)方法，发送Basic.Get请求。
    3. 调用k.getReply()方法，等待响应并返回。
  1.    private AMQCommand privateRpc(Method m) throws IOException, ShutdownSignalException{
  2.  
  3.     SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation(m);
  4.  
  5.     rpc(m, k);	// 发送请求
  6.  
  7.     if(_rpcTimeout == NO_RPC_TIMEOUT) {
  8.  
  9.         return k.getReply();	// 等待响应
  10.  
  11.     } else {
  12.  
  13.         try {
  14.  
  15.             return k.getReply(_rpcTimeout);
  16.  
  17.         } catch (TimeoutException e) {
  18.  
  19.             throw wrapTimeoutException(m, e);
  20.  
  21.         }
  22.  
  23.     }
  24.  
  25. }
  26.  
  27.  
  28.  
  29. public void rpc(Method m, RpcContinuation k) throws IOException {
  30.  
  31.     synchronized (_channelMutex) {
  32.  
  33.         ensureIsOpen();
  34.  
  35.         quiescingRpc(m, k);
  36.  
  37.     }
  38.  
  39. }
  40.  
  41.  
  42.  
  43. public void quiescingRpc(Method m, RpcContinuation k) throws IOException {
  44.  
  45.     synchronized (_channelMutex) {
  46.  
  47.         enqueueRpc(k);
  48.  
  49.         quiescingTransmit(m);
  50.  
  51.     }
  52.  
  53. }
  54.  
  55.  
  56.  
  57. public void enqueueRpc(RpcContinuation k) {
  58.  
  59.     doEnqueueRpc(() -> new RpcContinuationRpcWrapper(k));
  60.  
  61. }
  62.  
  63.  
  64.  
  65. public void quiescingTransmit(Method m) throws IOException {
  66.  
  67.     synchronized (_channelMutex) {
  68.  
  69.         quiescingTransmit(new AMQCommand(m));
  70.  
  71.     }
  72.  
  73. }
  74.  
  75.  
  76.  
  77. public void quiescingTransmit(AMQCommand c) throws IOException {
  78.  
  79.     synchronized (_channelMutex) {
  80.  
  81.         if (c.getMethod().hasContent()) {
  82.  
  83.            while (_blockContent) {
  84.  
  85.                 try {
  86.  
  87.                     _channelMutex.wait();
  88.  
  89.                 } catch (InterruptedException ignored) {
  90.  
  91.                     Thread.currentThread().interrupt();
  92.  
  93.                 }
  94.  
  95.                 ensureIsOpen();
  96.  
  97.             }
  98.  
  99.         }
  100.  
  101.         this._trafficListener.write(c);
  102.  
  103.         c.transmit(this);
  104.  
  105.     }
  106.  
  107. }

  1. 最终，在AMQCommand中执行transmit()方法，核心步骤如下：

    1. 获取AMQConnection对象。
    2. 分别将AMQContentHeader、Method和List<byte[]>对象转换成Frame对象。
    3. 通过AMQConnection对象发送数据。
  1.    public void transmit(AMQChannel channel) throws IOException {
  2.  
  3.     int channelNumber = channel.getChannelNumber();
  4.  
  5.     AMQConnection connection = channel.getConnection();
  6.  
  7.     synchronized (assembler) {
  8.  
  9.         Method m = this.assembler.getMethod();
  10.  
  11.         if (m.hasContent()) {
  12.  
  13.             byte[] body = this.assembler.getContentBody();
  14.  
  15.             Frame headerFrame = this.assembler.getContentHeader().toFrame(channelNumber, body.length);
  16.  
  17.             int frameMax = connection.getFrameMax();
  18.  
  19.             boolean cappedFrameMax = frameMax > 0;
  20.  
  21.             int bodyPayloadMax = cappedFrameMax ? frameMax - EMPTY_FRAME_SIZE : body.length;
  22.  
  23.  
  24.  
  25.             if (cappedFrameMax && headerFrame.size() > frameMax) {
  26.  
  27.                 String msg = String.format("Content headers exceeded max frame size: %d > %d", headerFrame.size(), frameMax);
  28.  
  29.                 throw new IllegalArgumentException(msg);
  30.  
  31.             }
  32.  
  33.             connection.writeFrame(m.toFrame(channelNumber));
  34.  
  35.             connection.writeFrame(headerFrame);
  36.  
  37.  
  38.  
  39.             for (int offset = 0; offset < body.length; offset += bodyPayloadMax) {
  40.  
  41.                 int remaining = body.length - offset;
  42.  
  43.  
  44.  
  45.                 int fragmentLength = (remaining < bodyPayloadMax) ? remaining
  46.  
  47.                         : bodyPayloadMax;
  48.  
  49.                 Frame frame = Frame.fromBodyFragment(channelNumber, body,
  50.  
  51.                         offset, fragmentLength);
  52.  
  53.                 connection.writeFrame(frame);
  54.  
  55.             }
  56.  
  57.         } else {
  58.  
  59.             connection.writeFrame(m.toFrame(channelNumber));
  60.  
  61.         }
  62.  
  63.     }
  64.  
  65.     connection.flush();
  66.  
  67. }

4 AMQCommand#

com.rabbitmq.client.impl.AMQCommand类实现了com.rabbitmq.client.Command接口，其成员变量CommandAssembler对象是AMQP规范中method、header和body的容器。

AMQCommand中提供了一个十分重要的方法：transmit(AMQChannel)。调用该方法能够将method、header和body通过Connection发送给RabbitMQ服务器。该方法在前几个小节都有介绍，核心步骤如下：

  1. 获取AMQConnection对象。
  2. 分别将AMQContentHeader、Method和List<byte[]>对象转换成Frame对象。
  3. 通过AMQConnection对象发送数据。
  1. public void transmit(AMQChannel channel) throws IOException {
  2.  
  3.     int channelNumber = channel.getChannelNumber();
  4.  
  5.     AMQConnection connection = channel.getConnection();
  6.  
  7.  
  8.  
  9.     synchronized (assembler) {
  10.  
  11.         Method m = this.assembler.getMethod();
  12.  
  13.         if (m.hasContent()) {
  14.  
  15.             byte[] body = this.assembler.getContentBody();
  16.  
  17.             Frame headerFrame = this.assembler.getContentHeader().toFrame(channelNumber, body.length);
  18.  
  19.             int frameMax = connection.getFrameMax();
  20.  
  21.             boolean cappedFrameMax = frameMax > 0;
  22.  
  23.             int bodyPayloadMax = cappedFrameMax ? frameMax - EMPTY_FRAME_SIZE : body.length;
  24.  
  25.             if (cappedFrameMax && headerFrame.size() > frameMax) {
  26.  
  27.                 String msg = String.format("Content headers exceeded max frame size: %d > %d", headerFrame.size(), frameMax);
  28.  
  29.                 throw new IllegalArgumentException(msg);
  30.  
  31.             }
  32.  
  33.             connection.writeFrame(m.toFrame(channelNumber));
  34.  
  35.             connection.writeFrame(headerFrame);
  36.  
  37.             for (int offset = 0; offset < body.length; offset += bodyPayloadMax) {
  38.  
  39.                 int remaining = body.length - offset;
  40.  
  41.                 int fragmentLength = (remaining < bodyPayloadMax) ? remaining
  42.  
  43.                         : bodyPayloadMax;
  44.  
  45.                 Frame frame = Frame.fromBodyFragment(channelNumber, body,
  46.  
  47.                         offset, fragmentLength);
  48.  
  49.                 connection.writeFrame(frame);
  50.  
  51.             }
  52.  
  53.         } else {
  54.  
  55.             connection.writeFrame(m.toFrame(channelNumber));
  56.  
  57.         }
  58.  
  59.     }
  60.  
  61.     connection.flush();
  62.  
  63. }

4.1 CommandAssembler#

com.rabbitmq.client.impl.CommandAssembler类中封装了AMQP规范中method、header和body。

4.1.1 Method#

com.rabbitmq.client.impl.Method抽象类代表AMQP规范中method，我们平常所使用的com.rabbitmq.client.AMQP接口中的Connection、Channel、Access、Exchange、Queue、Basic、Tx和Confirm等内部类都实现了该抽象类。

我们调用channel.basicPublish()等方法向RabbitMQ服务器发送消息，或者从通过注册Consumer监听RabbitMQ服务器的消息时，都会将method数据段转换成Method实现类进行处理。

Method.toFrame()方法则能将自己转换成Frame对象，进行发送。

4.1.2 AMQContentHeader#

com.rabbitmq.client.impl.AMQContentHeader抽象类代表AMQP规范中header。channel.basicPublish()方法形参BasicProperties实现了该抽象类，我们可以通过该对象为消息设置属性。

我们调用channel.basicPublish()等方法向RabbitMQ服务器发送消息，或者从通过注册Consumer监听RabbitMQ服务器的消息时，都会将method数据段转换成AMQContentHeader实现类进行处理。

AMQContentHeader.toFrame()方法则能将自己转换成Frame对象，进行发送。

5 Frame#

com.rabbitmq.client.impl.Frame代表AMQP wire-protocol frame（帧），主要包含以下成员变量：

  • type：帧类型。
  • channel：所属通道。
  • payload：输入载荷。
  • accumulator：输出载荷。

Frame还提供了静态方法readFrom()，可以从输入流中读取到Frame对象，主要提供给FrameHandler.readFrame()方法调用：

  1. public static Frame readFrom(DataInputStream is) throws IOException {
  2.  
  3.     int type;
  4.  
  5.     int channel;
  6.  
  7.     try {
  8.  
  9.         type = is.readUnsignedByte();
  10.  
  11.     } catch (SocketTimeoutException ste) {
  12.  
  13.         return null; // failed
  14.  
  15.     }
  16.  
  17.     if (type == 'A') {
  18.  
  19.         protocolVersionMismatch(is);
  20.  
  21.     }
  22.  
  23.     channel = is.readUnsignedShort();
  24.  
  25.     int payloadSize = is.readInt();
  26.  
  27.     byte[] payload = new byte[payloadSize];
  28.  
  29.     is.readFully(payload);
  30.  
  31.     int frameEndMarker = is.readUnsignedByte();
  32.  
  33.     if (frameEndMarker != AMQP.FRAME_END) {
  34.  
  35.         throw new MalformedFrameException("Bad frame end marker: " + frameEndMarker);
  36.  
  37.     }
  38.  
  39.     return new Frame(type, channel, payload);
  40.  
  41. }

 分类: RabbitMQ

 标签: RabbitMQ

来源：[RabbitMQ]Java客户端：源码概览 - Xianhuii - 博客园

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/384391
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号