使用 AMQ JMS 客户端

AMQ JMS 支持以下行业认可的标准和网络协议：

有关 AMQ JMS 支持的配置 的当前信息，请参阅红帽客户门户网站上的 Red Hat AMQ 支持的配置。

本节介绍核心 API 实体，并描述它们如何一起工作。

AMQ JMS 发送并 接收消息。使用消息 生产者和消费者 在连接的对等点之间传输消息。生产者和消费者通过会话建立 。会话通过 连接建立。连接由 连接工厂 创建。

发送对等点会创建一个制作者来发送消息。制作者有一个 目的地，用于标识远程对等点上的目标队列或主题。接收对等点会创建一个消费者来接收消息。与生产者一样，使用者也有一个目的地，用于标识远程对等点上的源队列或主题。

目标是 队列或主题 。在 JMS 中，队列和主题是包含消息的命名代理实体的客户端表示。

队列实施点对点语义。每条消息仅由一个消费者查看，并且消息在读取后从队列中删除。主题实施发布订阅语义。每个消息都会看到多个消费者，在读取后，消息仍可供其他消费者使用。

如需更多信息，请参阅 JMS 指南。

sudo 命令

在本文档中，sudo 用于任何需要 root 特权的命令。使用 sudo 时请谨慎操作，因为任何更改都可能会影响整个系统。有关 sudo 的更多信息，请参阅使用 sudo 命令。

文件路径

在本文档中，所有文件路径都对 Linux、UNIX 和类似操作系统（例如 /home/andrea）有效。在 Microsoft Windows 上，您必须使用对应的 Windows 路径（例如 C:\Users\andrea）。

变量文本

本文档包含代码块，其中的变量必须替换为特定于您的环境的值。变量文本以箭头括号括起，样式为方便的 monospace。例如，使用以下命令将 < project-dir&gt; 替换为环境的值：

$ cd <project-dir>

配置 Maven 环境，以从 Red Hat Maven 存储库下载客户端库。

客户端现在在 Maven 项目中可用。

作为在线存储库的替代选择，AMQ JMS 可以作为基于文件的 Maven 存储库安装到本地文件系统。

在此文档中，生成的本地目录被称为 &lt ;source-dir >。

Hello World 示例创建了一个与代理的连接，发送一条消息，其中包含到队列 队列的 问候消息，然后接收它。成功时，它会将收到的消息输出到控制台。

例如，在 Linux 上运行它会产生以下输出：

$ java -cp "target/classes/:target/dependency/*" org.apache.qpid.jms.example.HelloWorld
Hello world!

示例的源代码位于 < source-dir> /qpid-jms-examples/src/main/java 目录中。JNDI 和日志记录配置位于 < source-dir>/qpid-jms-examples/src/main/resources 目录中。

JMS 应用使用 InitialContext Factory 获取的 JNDI InitialContext 对象来查找 JMS 对象，如连接工厂。AMQ JMS 在 org.apache.qpid.jms.jndi.JmsInitialContextFactory 类中提供了 InitialContextFactory 的实施。

当 InitialContext 对象实例化时，会发现 InitialContextFactory 实现：

javax.naming.Context context = new javax.naming.InitialContext();

若要查找实施，环境中必须配置 JNDI。实现这一点的方法有三种：使用 jndi.properties 文件、使用系统属性或使用初始上下文 API。

使用 jndi.properties 文件

创建名为 jndi.properties 的文件，并将其放在 Java classpath 中。使用键 java.naming.factory.initial 添加属性。

java.naming.factory.initial = org.apache.qpid.jms.jndi.JmsInitialContextFactory

在基于 Maven 的项目中，jndi.properties 文件放置在 < project-dir> /src/main/resources 目录中。

使用系统属性

设置 java.naming.factory.initial 系统属性。

$ java -Djava.naming.factory.initial=org.apache.qpid.jms.jndi.JmsInitialContextFactory ...

使用初始上下文 API

使用 JNDI 初始上下文 API 以编程方式设置属性。

Hashtable<Object, Object> env = new Hashtable<>();

env.put("java.naming.factory.initial", "org.apache.qpid.jms.jndi.JmsInitialContextFactory");

InitialContext context = new InitialContext(env);

请注意，您可以使用相同的 API 为连接工厂、队列和主题设置 JNDI 属性。

JMS 连接工厂是创建连接的入口点。它使用一个连接 URI，对特定于应用程序的配置设置进行编码。

要设置工厂名称和连接 URI，请使用以下格式创建一个属性。您可以将此配置存储在 jndi.properties 文件中，或者设置对应的系统属性。

connectionFactory.<lookup-name> = <connection-uri>

例如，这是如何配置名为 app1 的工厂：

connectionFactory.app1 = amqp://example.net:5672?jms.clientID=backend

然后，您可以使用 JNDI 上下文使用名称 app1 来查找您配置的连接工厂：

ConnectionFactory factory = (ConnectionFactory) context.lookup("app1");

连接使用连接 URI 进行配置。连接 URI 指定远程主机、端口和一组配置选项，它们被设置为查询参数。有关可用选项的详情请参考 第 5 章 配置选项。

<scheme>://<host>:<port>[?<option>=<value>[&<option>=<value>...]]

对于未加密的连接，方案是 amqp ，用于 SSL/TLS 连接。

例如，以下是一个连接 URI，它连接到位于端口 5672 的主机 example.net，并将客户端 ID 设置为 后端 ：

amqp://example.net:5672?jms.clientID=backend

故障转移 URI

配置故障转移时，如果与当前服务器的连接丢失，客户端可以自动重新连接到其他服务器。故障转移 URI 具有前缀 故障切换： 在括号内包含以逗号分隔的连接 URI 列表。其他选项在末尾指定。

failover:(<connection-uri>[,<connection-uri>...])[?<option>=<value>[&<option>=<value>...]]

例如，以下是故障转移 URI，它可以连接到两个主机 host1 或 host2 之一：

failover:(amqp://host1:5672,amqp://host2:5672)?jms.clientID=backend

与连接 URI 示例一样，客户端可以使用故障转移配置中的 URI 配置多个不同的设置。这些设置在 第 5 章 配置选项 中详细介绍，其中 第 5.5 节 “故障转移选项” 部分是特别关注的部分。

SSL/TLS 服务器名称代表

当使用 amqps 方案指定 SSL/TLS 连接时，JVM 的 TLS 服务器名称调用(SNI)扩展中的主机片段可用于 TLS 握手期间通信所需的服务器主机名。如果指定了完全限定域名（如 "myhost.mydomain"），但没有使用非限定名称（如 "myhost"）或裸机 IP 地址，则会自动包含 SNI 扩展。

JMS 提供了使用 JNDI 查找特定于部署的队列和主题资源的 选项。

若要在 JNDI 中设置队列和主题名称，请以以下格式创建属性：将此配置放在 jndi.properties 文件中，或者设置对应的系统属性。

queue.<lookup-name> = <queue-name>
topic.<lookup-name> = <topic-name>

例如，以下属性定义了两个部署特定资源的名称 作业和 通知 ：

queue.jobs = app1/work-items
topic.notifications = app1/updates

然后，您可以通过其 JNDI 名称查找资源：

Queue queue = (Queue) context.lookup("jobs");
Topic topic = (Topic) context.lookup("notifications");

JNDI 属性值可以包含格式为 ${ <variable-name> } 的变量。该程序库通过搜索以下位置的顺序来解析变量值：

例如，在 Linux ${HOME} 上，解析为 HOME 环境变量，当前用户的主目录。

可以使用语法 ${ <variable-name> :- < default-value> } 提供默认值。如果没有找到 &lt ;variable-name > 的值，则使用默认值。

这些选项控制 JMS 对象的行为，如 Connection,Session,MessageConsumer, 和 MessageProducer。

预抓取策略选项

prefetch 策略决定每个 MessageConsumer 从远程对等点获取的消息数量，并保存在本地"prefetch"缓冲区中。

prefetch 的值可能会影响到队列或共享订阅上的多个消费者的信息分布。较高的值可导致一次向每个消费者发送更大的批处理。要实现更多循环分发，请使用较低的值。

重新发送策略选项

重新发送策略控制如何在客户端上处理重新设计的消息。

消息 ID 策略选项

消息 ID 策略控制分配给从客户端发送的消息 ID 的数据类型。

Presettle 策略选项

Presettle 策略控制生产者或消费者实例何时配置为使用 AMQP 预先设置的消息传递语义。

解序列化策略选项

解序列化策略提供了一种方式，控制哪些 Java 类型可信任从对象流中反序列化，同时从由序列化 Java 对象内容组成的传入 Object Message 检索正文。默认情况下，所有类型都会在尝试反序列化正文期间被信任。默认 deserialization 策略提供 URI 选项，允许指定白名单和 Java 类或软件包名称黑名单。

当使用普通 TCP 连接到远程服务器时，以下选项指定了底层套接字的行为。这些选项会与其他配置选项一起附加到连接 URI 中。

amqp://localhost:5672?jms.clientID=foo&transport.connectTimeout=30000

下面列出了完整的 TCP 传输选项。

SSL/TLS 传输通过使用 amqps URI 方案启用。因为 SSL/TLS 传输扩展了基于 TCP 的传输的功能，所以所有 TCP 传输选项都在 SSL/TLS 传输 URI 中有效。

amqps://myhost.mydomain:5671

下面列出了完整的 SSL/TLS 传输选项。

以下选项适用于与 AMQP 线协议相关的行为的各个方面。

故障转移 URI 从前缀 failover: 开始，并在括号内包含以逗号分隔的连接 URI 列表。其他选项在末尾指定。前缀为 jms. 的选项应用于括号之外的总体故障转移 URI，并影响 其生命周期的连接 对象。

failover:(amqp://host1:5672,amqp://host2:5672)?jms.clientID=foo&failover.maxReconnectAttempts=20

括号中的单个代理详情可以使用之前定义的 transport. 或 amqp. 选项。它们在每次主机都连接到时应用。

failover:(amqp://host1:5672?amqp.option=value,amqp://host2:5672?transport.option=value)?jms.clientID=foo

下面列出了用于故障转移的所有配置选项。

故障转移 URI 还支持定义嵌套选项，作为指定适用于所有嵌套代理 URI 的 AMQP 和传输选项值的方法。这使用相同的 传输实现。 和 amqp。之前为非故障切换代理 URI 概述的 URI 选项，但带有 failover.nested 的前缀。例如，要将 amqp.vhost 选项的值应用到连接到的每个代理，您可能有一个类似如下的 URI：

failover:(amqp://host1:5672,amqp://host2:5672)?jms.clientID=foo&failover.nested.amqp.vhost=myhost

客户端具有一个可选的发现模块，它提供自定义故障转移层，代理 URI 没有在初始 URI 中提供，而是通过与发现代理交互来发现。目前有两个发现代理实现：一个从文件中加载 URI 的文件监视器，以及一个多播监听程序，用于为侦听客户端广播其代理地址的 ActiveMQ 5.x 代理。

使用发现时，常规的与故障转移相关的选项集与之前详述的相同，主前缀从 故障转移 改为 发现。 以及用于提供对所有发现的代理 URI 通用的 嵌套 前缀。例如，如果没有代理 URI 详情，一般的发现 URI 可能类似如下：

discovery:(<agent-uri>)?discovery.maxReconnectAttempts=20&discovery.discovered.jms.clientID=foo

要使用文件观察器发现代理，请创建类似如下的代理 URI：

discovery:(file:///path/to/monitored-file?updateInterval=60000)

下面列出了文件观察器发现代理的 URI 选项。

要将多播发现代理与 ActiveMQ 5.x 代理搭配使用，请创建类似如下的代理 URI：

discovery:(multicast://default?group=default)

请注意，在上述多播代理 URI 中将 default 用作主机是一个特殊值，代理将替换为默认的 239.255.2.3:6155。您可以更改它，以指定用于多播配置的实际 IP 地址和端口。

下方列出了多播发现代理的 URI 选项。

使用 JMS 的应用通常使用 JNDI 获取应用所用的 ConnectionFactory 和 Destination 对象。这会使配置与程序分开，并使其与特定的客户端实施隔离。

为了使用这些示例，应将名为 jndi.properties 的文件放在类路径上，以配置 JNDI 上下文，如前面所述。

jndi.properties 文件的内容应与下面显示的内容匹配，它确定应使用客户端的 InitialContextFactory 实施，配置 ConnectionFactory 以连接到本地服务器，并定义名为 queue 的目标队列。

# Configure the InitialContextFactory class to use
java.naming.factory.initial = org.apache.qpid.jms.jndi.JmsInitialContextFactory

# Configure the ConnectionFactory
connectionfactory.myFactoryLookup = amqp://localhost:5672

# Configure the destination
queue.myDestinationLookup = queue

本例首先会创建一个 JNDI 上下文，使用它来查找 ConnectionFactory 和 Destination，使用工厂创建并启动 连接，然后创建一个 Session。然后，将创建一个 MessageProducer 到 Destination，并使用它发送一条消息。然后，连接 关闭，程序会退出。

此 Sender 示例的一个可运行变体位于 &lt ;source-dir>/qpid-jms-examples 目录中，以及之前在 第 3 章 开始使用 中涵盖的 Hello World 示例。

package org.jboss.amq.example;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;

public class Sender {
  public static void main(String[] args) throws Exception {
    try {
      Context context = new InitialContext(); 1

      ConnectionFactory factory = (ConnectionFactory) context.lookup("myFactoryLookup");
      Destination destination = (Destination) context.lookup("myDestinationLookup"); 2

      Connection connection = factory.createConnection("<username>", "<password>");
      connection.setExceptionListener(new MyExceptionListener());
      connection.start(); 3

      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 4

      MessageProducer messageProducer = session.createProducer(destination); 5

      TextMessage message = session.createTextMessage("Message Text!"); 6
      messageProducer.send(message, DeliveryMode.NON_PERSISTENT,
                           Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE); 7

      connection.close(); 8
    } catch (Exception exp) {
      System.out.println("Caught exception, exiting.");
      exp.printStackTrace(System.out);
      System.exit(1);
    }
  }

  private static class MyExceptionListener implements ExceptionListener {
    @Override
    public void onException(JMSException exception) {
      System.out.println("Connection ExceptionListener fired, exiting.");
      exception.printStackTrace(System.out);
      System.exit(1);
    }
  }
}

请注意，这只是一个示例。现实应用通常使用长期消息Producer，并随着时间的推移发送许多消息。打开然后关闭每个 消息 的连接、会话 和消息生产通常效率低下。

本例首先创建一个 JNDI 上下文，使用它来查找 ConnectionFactory 和 Destination，使用工厂创建和启动 连接，然后创建一个 Session。然后，为 Destination 创建一个 MessageConsumer，使用它收到一条消息，其内容会输出到控制台。然后，连接关闭，程序会退出。相同的 JNDI 配置与 发送示例 中相同。

此 Receiver 示例的一个可执行变体包含在客户端分发的示例目录中，以及之前在 第 3 章 开始使用 中涵盖的 Hello World 示例。

package org.jboss.amq.example;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;

public class Receiver {
  public static void main(String[] args) throws Exception {
    try {
      Context context = new InitialContext(); 1

      ConnectionFactory factory = (ConnectionFactory) context.lookup("myFactoryLookup");
      Destination destination = (Destination) context.lookup("myDestinationLookup"); 2

      Connection connection = factory.createConnection("<username>", "<password>");
      connection.setExceptionListener(new MyExceptionListener());
      connection.start(); 3

      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 4

      MessageConsumer messageConsumer = session.createConsumer(destination); 5

      Message message = messageConsumer.receive(5000); 6

      if (message == null) { 7
        System.out.println("A message was not received within given time.");
      } else {
        System.out.println("Received message: " + ((TextMessage) message).getText());
      }

      connection.close(); 8
    } catch (Exception exp) {
      System.out.println("Caught exception, exiting.");
      exp.printStackTrace(System.out);
      System.exit(1);
    }
 }

  private static class MyExceptionListener implements ExceptionListener {
    @Override
    public void onException(JMSException exception) {
      System.out.println("Connection ExceptionListener fired, exiting.");
      exception.printStackTrace(System.out);
      System.exit(1);
    }
  }
}

请注意，这只是一个示例。现实应用通常使用长期的 MessageConsumer，并随着时间的推移接收许多消息。打开然后关闭每个消息 的连接、会话 和 MessageConsumer 通常效率较低。

SSL/TLS 连接可以配置为使用原生 OpenSSL 实现来提高性能。要使用 OpenSSL，必须启用 transport.useOpenSSL 选项，并且必须在 classpath 上提供 OpenSSL 支持库。

要在 Red Hat Enterprise Linux 上使用系统安装的 OpenSSL 库，请安装 openssl 和 apr RPM 软件包，并在 POM 文件中添加以下依赖项：

<dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-tcnative</artifactId>
  <version>2.0.46.Final-redhat-00001</version>
  <classifier>linux-x86_64-fedora</classifier>
</dependency>

Netty 项目中提供了 OpenSSL 库实现列表。

当与适当配置的服务器一起使用时，客户端可以配置为使用 Kerberos 进行身份验证。要启用 Kerberos，请使用以下步骤。

使用的确切配置将取决于您如何为连接建立凭证，以及使用的特定 LoginModule。有关 Oracle Krb5LoginModule 的详情，请查看 Oracle Krb5LoginModule 类参考。有关 IBM Java 8 Krb5LoginModule 的详情，请查看 IBM Krb5LoginModule 类参考。

可以将 LoginModule 配置为建立用于 Kerberos 进程的凭证，如指定主体以及是否使用现有的票据缓存或 keytab。但是，如果 LoginModule 配置不提供建立所有必要的凭据的方法，则请求并从客户端连接对象传递用户名和密码值（如果使用 ConnectionFactory 选项创建连接或之前通过 URI 选项配置）。

请注意，Kerberos 仅支持验证目的。使用 SSL/TLS 连接进行加密。

以下连接 URI 选项可用于影响 Kerberos 身份验证过程。

与前面详述的 amqp. 和 transport. 选项类似，这些选项必须基于每个主机指定，或者作为故障切换 URI 中的 all-host 嵌套选项。

消息传递系统使用消息确认来跟踪消息的目标是否真正完成。

发送消息后，在邮件被发送并被确认前有一个时间段（消息为"in flight"）。如果此时丢失网络连接，消息交付的状态未知，发送可能需要在应用程序代码中特殊处理，以确保其完成。

以下小节描述了连接失败时消息交付的条件。

非转换生成者带有未确认的发送

如果消息在 flight 中，它会在重新连接后再次发送，只要未设置发送超时，且尚未设置。

不需要用户操作。

带有未提交事务的 transacted producer

如果消息在 flight 中，它会在重新连接后再次发送。如果发送是新事务中的第一个，则在重新连接后发送将继续正常。如果以前的事务中发送，则事务被视为失败，后续提交操作都会引发 TransactionRolledBackException。

为确保发送，用户必须重新发送属于失败事务的任何消息。

带有待处理提交的 transacted producer

如果提交处于 flight，则事务被视为失败，后续提交操作都会引发 TransactionRolledBackException。

为确保发送，用户必须重新发送属于失败事务的任何消息。

非转换的消费者，带有未确认的发送

如果收到消息但尚未确认，则确认消息不会产生任何错误，但不会给客户端产生任何操作。

由于收到的消息没有被确认，因此生成者可能会重新发送它。为避免重复，用户必须根据消息 ID 过滤出重复的消息。

带有未提交事务的 transacted consumer

如果还没有提交活动事务，它将被视为失败，并丢弃任何待处理的确认。任何后续提交操作都会抛出 TransactionRolledBackException。

生产者可能会重新发送属于事务的消息。为避免重复，用户必须根据消息 ID 过滤出重复的消息。

使用待处理的提交转换消费者

如果提交处于 flight，则事务将被视为失败。任何后续提交操作都会抛出 TransactionRolledBackException。

生产者可能会重新发送属于事务的消息。为避免重复，用户必须根据消息 ID 过滤出重复的消息。

除了 JMS 规范中定义的外，客户端支持另外两种会话确认模式。

个人确认

在这个模式中，应用程序必须使用会话处于 CLIENT_ACKNOWLEDGE 模式时使用的 Message.acknowledge （） 方法单独确认消息。与 CLIENT_ACKNOWLEDGE 模式不同，只有目标消息被确认。所有其他发送的消息仍未确认。用于激活此模式的整数值是 101。

connection.createSession(false, 101);

没有确认

在此模式中，在发送到客户端之前，服务器接受消息，并且客户端不会执行确认。客户端支持两个整数值来激活此模式，100 和 257。

connection.createSession(false, 100);

客户端使用 SLF4J API，允许用户根据自己的需要选择特定的日志记录实施。例如，用户可以提供 slf4j-log4j 绑定来选择 Log4J 实现。有关 SLF4J 的详情，请访问 其网站。

客户端使用位于 org.apache.qpid.jms 层次结构中的 Logger 名称，您可以使用它来根据您的需要配置日志实施。

调试时，从 Qpid Proton AMQP 1.0 库启用其他协议追踪日志有时非常有用。实现此目标的方法有两种。

您还可以将客户端配置为发出输入和输出字节的低级别追踪。若要启用此功能，请将选项 transport.traceBytes=true 添加到 连接 URI，并将 org.apache.qpid.jms.transports.netty.NettyTcpTransport 日志记录器配置为日志级别 DEBUG。

使用以下步骤在应用程序中启用追踪：

要查看应用程序捕获的 trace，请使用 Jaeger Getting Started 来运行 Jaeger 基础架构和控制台。

AMQP 消息 使用 AMQP 类型系统 组成。具有这种通用格式是以不同语言的 AMQP 客户端能够相互互操作的原因之一。本节旨在记录客户端与所使用的各种 JMS 消息类型相关的 AMQP 有效负载和接收的 AMQP 有效负载，以帮助将客户端与其他 AMQP 客户端一起使用。

AMQ Broker 旨在与 AMQP 1.0 客户端互操作。检查以下内容以确保为 AMQP 消息传递配置了代理：