使用 AMQ 核心协议 JMS 客户端

AMQ 核心协议 JMS 支持以下行业标准和网络协议：

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

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

AMQ 核心协议 JMS 发送并 接收消息。消息使用消息制作者 和消费者 在连接的对等点间传输。生产者和消费者是通过 会话 建立的。会话通过连接 建立。连接通过 连接工厂 创建。

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

目的地可以是 队列，也可以是 主题。在 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 存储库安装到本地文件系统中。

这个示例为名为 exampleQueue 的队列创建一个消费者和制作者。它发送文本消息，然后接收它，并将收到的消息打印到控制台。

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

$ java -cp "target/classes:target/dependency/*" org.apache.activemq.artemis.jms.example.QueueExample
Sent message: This is a text message
Received message: This is a text message

示例的源代码位于 < install-dir> /examples/features/standard/queue/src 目录中。< install-dir> /examples/features/standard 目录中提供了 其他示例。

JMS 应用使用从 InitialContext Factory 获取的 JNDI InitialContext 对象来查找 JMS 对象，如连接工厂。AMQ Core Protocol JMS 在 org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory 类中提供 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.activemq.artemis.jndi.ActiveMQInitialContextFactory

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

使用系统属性

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

$ java -Djava.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory ...

使用初始上下文 API

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

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

env.put("java.naming.factory.initial", "org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory");

InitialContext context = new InitialContext(env);

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

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

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

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

例如，您可以配置名为 app1 的工厂：

connectionFactory.app1 = tcp://example.net:61616?clientID=backend

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

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

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

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

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

tcp://example.net:61616?clientID=backend

除了 tcp 外，AMQ 核心协议 JMS 还支持 vm、udp 和 jgroups 方案。它们代表备用传输，并在代理上具有相应的 acceptor 配置。

故障转移 URI

URI 可以包含多个目标连接 URI。如果初始连接到一个目标失败，则尝试另一个目标。它们采用以下格式：

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

括号外的选项适用于所有连接 URI。

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

如需更多信息，请参阅 第 8 章 流控制。

客户端可以通过为属性 minLargeMessageSize 设置值来启用大型消息支持。任何大于 minLargeMessageSize 的消息都被视为大型消息。

客户端可以接收有关所有主和从属代理的信息，以便在连接失败时，它可以重新连接到从代理。然后，从代理会自动重新创建每个连接上存在的会话和消费者，然后再故障转移。此功能可以在应用程序中手动重新连接逻辑。

在从中重新创建会话时，它不知道已经发送或确认的消息。故障转移时任何动态发送或确认都可能会丢失。但是，即使没有透明故障转移，也很简单地保证 一次和一次 交付，即使在出现重复检测和重试事务时也是如此。

当客户端在可配置的时间段内没有从代理接收数据包时，客户端会检测连接失败。请参阅 第 6.3 节 “检测死连接” 了解更多信息。

您可以通过多种方式配置客户端来接收关于主设备和从系统的信息。一种选择将客户端配置为连接到特定代理，然后接收集群中其他代理的信息。请参阅 第 6.7 节 “配置静态发现” 了解更多信息。但是，最常见的方法是使用代理 发现。有关如何配置代理发现的详情，请参考 第 6.6 节 “配置动态发现”。

另外，您可以通过在用于连接代理的 URI 的查询字符串中添加参数来配置客户端，如下例所示。

connectionFactory.ConnectionFactory=tcp://localhost:61616?ha=true&reconnectAttempts=3

ConnectionFactory cf =  ActiveMQJMSClient.createConnectionFactory(...)
cf.setInitialConnectAttempts(3);

<configuration>
 <core>
  ...
  <initial-connect-attempts>3</initial-connect-attempts> 1
  ...
 </core>
</configuration>

在某些情况下，您可能不希望自动客户端故障转移，但更喜欢在失败处理程序中对您自己的重新连接逻辑进行编码。这称为 应用程序级别的 故障切换，因为故障转移是在应用程序级别处理的。

要在使用 JMS 时实施应用级别的故障切换，请在 JMS 连接上设置 ExceptionListener 类。在检测到连接失败时，代理会调用 ExceptionListener。在 ExceptionListener 中，您应该关闭旧的 JMS 连接。您可能还希望从 JNDI 查找新的连接工厂实例并创建新的连接。

只要它从代理接收数据，客户端会认为连接处于活动状态。通过为 client-failure-check-period 属性提供值，将客户端配置为检查其连接是否有失败。网络连接的默认检查周期为 30,000 毫秒或 30 秒，而 in-VM 连接的默认值为 -1，这意味着，如果没有接收数据，客户端永远不会从侧失败连接。

通常，您将检查周期设置为小于代理连接时间到实时的值，这样可确保在出现临时故障时客户端可以重新连接。

为检测死连接设置检查周期

以下示例演示了如何将检查周期设置为 10,000 毫秒。

默认情况下，客户端可以为自己的连接设置生存时(TTL)。以下示例演示了如何设置 TTL。

客户端应用程序必须在退出前以受控的方式关闭其资源，以防止发生死连接。在 Java 中，建议在 finally 块中关闭连接：

Connection jmsConnection = null;
try {
   ConnectionFactory jmsConnectionFactory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(...);
   jmsConnection = jmsConnectionFactory.createConnection();
   ...use the connection...
}
finally {
   if (jmsConnection != null) {
      jmsConnection.close();
   }
}

您可以配置 AMQ 核心协议 JMS，以便在尝试建立连接时发现代理列表。

如果您在客户端上使用 JNDI 查找 JMS 连接工厂实例，您可以在 JNDI 上下文环境中指定这些参数。通常，参数在名为 jndi.properties 的文件中定义。连接工厂的 URI 中的主机和部分应与代理 broker.xml 配置文件中的对应 broadcast-group 中的 group-address 和 group-port 匹配。以下是配置为连接到代理的发现组的 jndi.properties 文件示例。

java.naming.factory.initial = ActiveMQInitialContextFactory
connectionFactory.myConnectionFactory=udp://231.7.7.7:9876

当由客户端应用程序从 JNDI 下载此连接工厂并从其创建 JMS 连接时，这些连接将通过侦听代理发现组配置中指定的多播地址来在维护的服务器列表中进行负载平衡。

作为使用 JNDI 的替代选择，您可以在创建 JMS 连接工厂时直接在 Java 代码中指定发现组参数。以下代码提供了如何执行此操作的示例。

final String groupAddress = "231.7.7.7";
final int groupPort = 9876;

DiscoveryGroupConfiguration discoveryGroupConfiguration = new DiscoveryGroupConfiguration();
UDPBroadcastEndpointFactory udpBroadcastEndpointFactory = new UDPBroadcastEndpointFactory();
udpBroadcastEndpointFactory.setGroupAddress(groupAddress).setGroupPort(groupPort);
discoveryGroupConfiguration.setBroadcastEndpointFactory(udpBroadcastEndpointFactory);

ConnectionFactory jmsConnectionFactory = ActiveMQJMSClient.createConnectionFactoryWithHA
    (discoveryGroupConfiguration, JMSFactoryType.CF);

Connection jmsConnection1 = jmsConnectionFactory.createConnection();
Connection jmsConnection2 = jmsConnectionFactory.createConnection();

可以使用 setter 方法 setRefreshTimeout （） 在 DiscoveryGroupConfiguration 上直接设置刷新超时。默认值为 10000 毫秒。

在首次使用时，连接工厂将确保它在创建第一个连接前等待这个时间。默认等待时间为 10000 毫秒，但您可以通过将新值传递给 DiscoveryGroupConfiguration.setDiscoveryInitialWaitTimeout （） 来更改它。

有时，您可能无法在您要使用的网络中使用 UDP。在这种情况下，您可以使用可能服务器的初始列表配置连接。列表只能是一个您知道始终可用的代理，或者至少有一个代理列表可用。

这并不意味着您必须知道所有服务器需要托管的位置。您可以将这些服务器配置为使用可靠的服务器进行连接。连接后，其连接详情将从服务器传播到客户端。

如果您在客户端上使用 JNDI 查找 JMS 连接工厂实例，您可以在 JNDI 上下文环境中指定这些参数。通常，参数在名为 jndi.properties 的文件中定义。以下是一个 jndi.properties 文件示例，它提供了一个静态代理列表，而不使用动态发现。

java.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory
connectionFactory.myConnectionFactory=(tcp://myhost:61616,tcp://myhost2:61616)

当客户端使用上述连接工厂时，其连接将在括号中定义的代理列表中进行负载平衡（）。

如果您直接实例化 JMS 连接工厂，您可以在创建 JMS 连接工厂时明确指定连接器列表，如下例所示。

HashMap<String, Object> map = new HashMap<String, Object>();
map.put("host", "myhost");
map.put("port", "61616");
TransportConfiguration broker1 = new TransportConfiguration
    (NettyConnectorFactory.class.getName(), map);

HashMap<String, Object> map2 = new HashMap<String, Object>();
map2.put("host", "myhost2");
map2.put("port", "61617");
TransportConfiguration broker2 = new TransportConfiguration
    (NettyConnectorFactory.class.getName(), map2);

ActiveMQConnectionFactory cf = ActiveMQJMSClient.createConnectionFactoryWithHA
    (JMSFactoryType.CF, broker1, broker2);

连接器定义客户端如何连接到代理。您可以使用 JMS 连接工厂从客户端配置它们。

Map<String, Object> connectionParams = new HashMap<String, Object>();

connectionParams.put(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, 61617);

TransportConfiguration transportConfiguration =
    new TransportConfiguration(
    "org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory", connectionParams);

ConnectionFactory connectionFactory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, transportConfiguration);

Connection jmsConnection = connectionFactory.createConnection();

要写入大型消息，请使用 BytesMessage.writeBytes （） 方法。以下示例从文件中读取字节并将其写入消息：

BytesMessage message = session.createBytesMessage();
File inputFile = new File(inputFilePath);
InputStream inputStream = new FileInputStream(inputFile);

int numRead;
byte[] buffer = new byte[1024];

while ((numRead = inputStream.read(buffer, 0, buffer.length)) != -1) {
    message.writeBytes(buffer, 0, numRead);
}

要从大型消息中读取，请使用 BytesMessage.readBytes （） 方法。以下示例从消息读取字节并将其写入文件中：

BytesMessage message = (BytesMessage) consumer.receive();
File outputFile = new File(outputFilePath);
OutputStream outputStream = new FileOutputStream(outputFile);

int numRead;
byte buffer[] = new byte[1024];

for (int pos = 0; pos < message.getBodyLength(); pos += buffer.length) {
    numRead = message.readBytes(buffer);
    outputStream.write(buffer, 0, numRead);
}

消息组是具有以下特征的消息集：

如果您希望同一消费者按顺序处理属性的特定值的所有消息时，消息组很有用。例如，您可能希望将任何特定库存购买的订购由同一消费者按顺序处理。为此，您可以创建用户池，然后将库存名称设置为 message 属性的值。这样可确保特定库存的所有消息始终由同一消费者处理。

设置组 ID

以下示例演示了如何将消息组与 AMQ 核心协议 JMS 搭配使用。

其他资源

如需了解如何配置和使用消息组的工作示例，请参阅 < install-dir> /examples/features/standard 下的 message-group 和 message-group2。

AMQ Broker 包括自动重复消息检测，它过滤出它收到的任何重复消息，因此您不必对您自己的重复检测逻辑进行编码。

若要启用重复消息检测，请为消息属性 _AMQ_DUPL_ID 提供唯一的值。代理收到消息时，它会检查 _AMQ_DUPL_ID 是否具有值。如果存在，代理会在其内存缓存中检查，以查看它是否已收到带有该值的消息。如果找到具有相同值的消息，则传入的消息将被忽略。

如果您要在事务中发送消息，则不必为事务中的每个消息设置 _AMQ_DUPL_ID，而是仅在其中之一。如果代理检测到事务中任何消息的重复消息，它会忽略整个事务。

设置重复的 ID 消息属性

以下示例演示了如何使用 AMQ 核心协议 JMS 设置重复的检测属性。请注意，为方便起见，客户端使用常量 org.apache.activemq.artemis.api.core.Message.HDR_DUPLICATE_DETECTION_ID 作为重复 ID 属性 _AMQ_DUPL_ID 的名称。

Message jmsMessage = session.createMessage();
String myUniqueID = "This is my unique id";
message.setStringProperty(HDR_DUPLICATE_DETECTION_ID.toString(), myUniqueID);

通过 AMQ Core Protocol JMS，您可以截获进入或退出客户端的数据包，允许您审核数据包或过滤消息。拦截器可以更改它们拦截器的数据包。这使得拦截器强大，但也是您应该谨慎使用的功能。

拦截器必须实施 intercept （） 方法，它返回一个 布尔值。如果返回的值为 true，则消息数据包将继续。如果返回的值为 false，该过程将中止，则不会调用其他拦截器，并且消息数据包不会被进一步处理。

消息拦截器对主客户端代码透明发生，除了在阻塞发送模式下发送传出数据包时。当启用了阻塞而发送传出数据包并且数据包遇到返回 false 的拦截器时，会抛出 ActiveMQException。thrown 异常包含拦截器的名称。

您的拦截器必须实施 org.apache.artemis.activemq.api.core.Interceptor 接口。客户端拦截器类及其依赖项必须添加到客户端的 Java 类路径中，才能正确实例化和调用。

package com.example;

import org.apache.artemis.activemq.api.core.Interceptor;
import org.apache.activemq.artemis.core.protocol.core.Packet;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;

public class MyInterceptor implements Interceptor {
    private final int ACCEPTABLE_SIZE = 1024;

    @Override
    boolean intercept(Packet packet, RemotingConnection connection) throws ActiveMQException {
        int size = packet.getPacketSize();
        if (size <= ACCEPTABLE_SIZE) {
            System.out.println("This Packet has an acceptable size.");
            return true;
        }
        return false;
    }
}

客户端缓冲区中保存的最大消息大小由窗口大小决定。AMQ 核心协议 JMS 的窗口的默认大小为 1 MiB，或 1024114 1024 字节。对于大多数用例，默认值是正常的。对于其他情况，查找窗口大小的最佳值可能需要基准测试您的系统。如果需要更改默认，AMQ Core Protocol JMS 允许您设置缓冲区窗口大小。

以下示例演示了如何在使用 AMQ Core Protocol JMS 时设置 consumer window size 参数。每个示例将消费者窗口大小设置为 300,000 字节。

窗口大小按信信在代理和生成者之间协商，每个字节在窗口中一个信用卡。当消息被发送和使用后，生产者必须请求并被授予来自代理的信信，然后才能发送更多消息。生产者和代理之间的信信交换规定了它们之间的数据流。

以下示例演示了如何在使用 AMQ 核心协议 JMS 时将制作者窗口大小设置为 1024 字节。

快速消费者可以在信息消耗时更快地处理消息。如果您认为消息传递系统中的消费者非常快，请考虑将窗口大小设置为 -1。将窗口大小设置为此值允许客户端未绑定消息缓冲。但是请谨慎使用此设置。如果消费者无法在收到消息时快速处理消息，客户端的内存可能会溢出。

为快速消费者设置窗口大小

以下示例演示了如何在使用 AMQ Core Protocol JMS 客户端（快速消费者）时将窗口大小设置为 -1。

减慢消费者需要大量时间来处理每个消息。在这些情况下，不建议在客户端上缓冲消息。代理中保留消息以供其他用户使用。关闭缓冲区的一个优点是，它在队列的多个消费者之间提供确定性分布。要通过禁用客户端缓冲来处理较慢的用户，请将窗口大小设置为 0。

为较慢的用户设置窗口大小

以下示例演示了如何在使用 AMQ Core Protocol JMS 客户端（消息较慢）时将窗口大小设置为 0。

其他资源

如需示例，请参阅 < install-dir> /examples/standard 中的 no-consumer-buffering 示例，其中演示了如何配置代理以防止在处理缓慢消费者时消费者缓冲。

您可以规范消费者可以使用消息的速率。也称为 节流，可以规范消耗率，确保消费者不会以比配置允许的速度快地消耗消息。

速率必须是正整数才能启用此功能，它是以每秒消息单位指定的最大所需消息消耗率。将速率设置为 -1 可禁用速率限制的流控制。默认值为 -1。

以下示例显示了一个客户端，它将消耗消息的速度限制为每秒 10 个消息。

其他资源

有关如何限制消费者率的工作示例，请参阅 & lt;install-dir> /examples/standard 中的 consumer-rate-limit 示例。

AMQ 核心协议 JMS 也可以限制制作者发送消息的速率。生成者率以每秒消息为单位指定。将它设置为 -1 （默认）会禁用速率限制的流控制。

以下示例演示了如何在制作者使用 AMQ Core Protocol JMS 时设置发送消息的速度。每个示例将最大率设置为每秒 10 个消息。

其他资源

有关如何限制发送消息率的工作示例，请参阅 <install-dir> /examples/standard 中的 producer-rate-limit 示例。