使用 AMQ JMS 客户端

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

有关 AMQ JMS 支持的配置，请参阅红帽客户门户网站中的 Red Hat AMQ 7 支持的配置。

本节介绍核心 API 实体，并描述它们如何协同运作。

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

发送对等点会创建一个制作者来发送消息。制作者具有在远程同级上标识目标队列或主题 的目的地。接收方创建接收消息的消费者。与制作者一样，消费者也有在远程同级上标识源队列或主题的目的地。

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

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

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

sudo 命令

在本文档中，sudo 用于任何需要 root 权限的命令。使用 sudo 时要小心，因为任何更改都可能会影响整个系统。有关 sudo 的详情请参考 使用 sudo 命令。

文件路径

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

变量文本

本文档包含代码块，它们需要使用特定于环境的值替换。变量文本括在箭头大括号内，样式为圆形单空间。例如，在以下命令中，将 <project-dir> 替换为您的环境的值：

$ cd <project-dir>

配置您的 Maven 环境，以从红帽 Maven 存储库下载客户端库。

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

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

在本文档中生成的本地目录被称为 <source-dir>。

Hello World 示例创建与代理的连接，发送一条包含问候语的消息到 queue 队列，然后重新接收它。成功后，它会将收到的消息打印到控制台。

例如，在 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 应用使用从 InitialContextFactory 获取的 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 类路径中。使用键 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 用于未加密的连接，amqps 用于 SSL/TLS 连接。

例如，以下连接 URI 通过端口 5672 连接到主机 example.net，并将客户端 ID 设置为 backend ：

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

故障切换 URI

配置故障转移后，客户端可以自动重新连接到另一服务器（如果与当前服务器的连接丢失）。故障切换 URI 具有前缀 failover:，它包含在括号内以逗号分隔的连接 URI 列表。末尾指定了附加选项。

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

例如，以下是可以连接到两个主机 host1 或 host2 的故障切换 URI：

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

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

SSL/TLS Server Name Indication

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

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

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

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

例如，以下属性为两个特定于部署的资源定义名称 jobs 和 notifications ：

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>} 提供默认值。如果没有找到 <variable-name> 的值，则会改为使用默认值。

这些选项控制 JMS 对象的产生，如 Connection、Session、MessageConsumer 和 MessageProducer。

预定义策略选项

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

prefetch 的值可能会影响在队列或共享订阅上向多个消费者发送消息。更高的值可能会导致向每位消费者同时发送的批处理。要实现更加均匀的循环分布，请使用更低的值。

重新传送策略选项

重新传送策略控制如何在客户端上处理重新传送消息。

消息 ID 策略选项

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

Presettle 策略选项

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

序列化策略选项

反序列化策略提供了控制 Java 类型被信任从对象流中取消序列化的方法，同时从传入的 ObjectMessage 检索由序列化 Java Object 内容组成的 正文。默认情况下，在尝试对正文进行反序列的过程中，所有类型都是受信任的。默认反序列策略提供 URI 选项，允许指定 Java 类或软件包名称的白名单和黑名单。

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

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

下方列出了一整套 TCP 传输选项：

使用 amqps URI 方案启用 SSL/TLS 传输。由于 SSL/TLS 传输扩展基于 TCP 的传输功能，因此所有 TCP 传输选项都在 SSL/TLS 传输 URI 上有效。

amqps://myhost.mydomain:5671

下方列出了完整的 SSL/TLS 传输选项集合。

以下选项适用于与 AMQP 线路协议相关的行为方面：

故障转移 URI 以前缀 failover: 开头，并在括号内包含一个用逗号分开的连接 URI 列表。末尾指定了附加选项。前缀为 jms. 的选项应用于括号之外的整体故障切换 URI，并影响其生命周期中的 Connection 对象。

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 还支持将嵌套选项定义为指定 AMQP 和传输选项值的方法，适用于所有嵌套代理 URI。这可以通过之前为非故障切换代理 URI 提供的相同 transport. 和 amqp. 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 代理配合使用，以广播其代理地址以供侦听客户端。

使用发现时的常规故障转移相关选项集与之前详述的相同，主前缀从 failover. 改为 discovery.，以及用于提供所有发现的代理 URI 的 URI 选项的 nested 前缀。例如，如果没有代理 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 Context，使用它查找 ConnectionFactory 和 Destination，使用工厂创建并启动 Connection，然后创建一个 Session。然后，MessageProducer 为 Destination 创建，并使用它发送信息。然后 Connection 被关闭，程序会退出。

这个 Sender 示例的可运行变体位于 <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);
    }
  }
}

请注意，这只是一个示例。现实应用通常会使用长期存在的 MessageProducer，并随着时间推移发送多个消息。打开并关闭每个信息 Connection、Session 和 MessageProducer 通常不高效。

这个示例首先创建一个 JNDI 上下文，使用它查找 ConnectionFactory 和 Destination，使用工厂创建并启动 Connection，然后创建一个 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，并随着时间推移接收许多使用它的信息。打开并关闭每个消息的 Connection、Session 和 MessageConsumer 通常并不高效。

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

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

<dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-tcnative</artifactId>
  <version>2.0.39.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 创建 Connection 或者之前通过其 URI 选项配置，则可以从客户端 Connection 对象请求和传递用户名和密码值。

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

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

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

消息传递系统使用消息确认来跟踪发送消息的目标是否真正实现。

发送邮件时，发送邮件之后和确认之前有一段时间（消息为"在机中"）。如果在这段时间内丢失网络连接，消息发送的状态未知，并且传输可能需要在应用代码中特殊处理以确保完成。

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

不经过翻译且交付未经确认的生产者

如果邮件处于运行状态，会在重新连接后再次发送，只要未设置发送超时且未通过。

不需要用户操作。

有未提交的事务的转译制作者

如果邮件处于转机状态，则在重新连接后再次发送。如果发送是新事务中的第一个，则在重新连接后发送会正常进行。如果之前发送了事务，则事务被视为失败，且任何后续的提交操作都会抛出 TransactionRolledBackException。

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

带有待处理的提交进行转换的制作者

如果提交处于机状态，则事务被视为失败，且任何后续的提交操作都会抛出 TransactionRolledBackException。

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

带有未经确认的交付的非已传输消费者

如果收到消息但尚未确认，则确认消息不会生成任何错误，但不会导致客户端不执行任何操作。

由于收到的消息未被确认，生产者可能会重新发送消息。为避免重复，用户必须通过消息 ID 过滤掉重复的消息。

带有未提交交易的转换消费者

如果一个活跃的事务尚未提交，它将被视为失败，任何待处理的确认都会被丢弃。任何后续提交操作都会抛出 TransactionRolledBackException。

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

带有待处理的提交进行转换的消费者

如果提交正在进行，则该交易将被视为失败。任何后续提交操作都会抛出 TransactionRolledBackException。

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

客户端支持除 JMS 规范中定义的两种其他会话确认模式：

个人确认

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

connection.createSession(false, 101);

Nofirm

在这种模式下，在将消息分配到客户端之前，服务器将接受消息，并且客户端不会执行任何确认。客户端支持两个整数值来激活此模式，即 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 客户端一起使用。

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

AMQ 互连可与任何 AMQP 1.0 客户端配合工作。检查以下内容以确保正确配置了组件：