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 存储库安装到您的本地文件系统中。

这个示例为名为 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 应用使用从 InitialContextFactory 获取的 JNDI InitialContext 对象来查找 JMS 对象，如连接工厂。AMQ 核心协议 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>...]]

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

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

除了 tcp 外，AMQ 核心协议 JMS 也支持 vm、udp 和 jgroups 方案。这些代表备用传输，并在代理中配置相应的接收器。

故障切换 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>

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

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 的消息都被视为一个大信息。

客户端可以接收所有 master 和从代理的信息，以便在连接失败时可以重新连接到从代理。然后从代理会在故障转移前自动重新创建每个连接上存在的任何会话和消费者。此功能使您不必在应用程序中手动编码重新连接逻辑。

在从卷上重新创建会话时，它不知道已经发送或确认的消息。故障切换时的任何内部发送或确认也可能丢失。但是，即便没有透明故障转移，也很容易通过将重复检测和重试事务结合使用，保证 一次并仅 一次交付一次（即便在故障的情况下也是如此）。

当客户端没有在可配置的时间段内从代理接收数据包时，客户端会检测到连接失败。如需更多信息，请参阅 第 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 的文件中定义。连接工厂的主机和部分应与代理 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);
}

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

如果您希望同一使用者以序列方式处理属性的特定值的所有消息，则消息组很有用。例如，您可能希望订单让同一消费者序列处理任何特定股票购买。为此，您可以创建一个消费者池，然后将库存名称设置为消息属性的值。这可确保特定库存的所有消息始终由同一消费者处理。

设置组 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 核心协议 JMS，您可以拦截进入或退出客户端的数据包，允许您审核数据包或过滤消息。拦截器可以更改它们拦截的数据包。这使得拦截器功能强大，同时也是您应该谨慎使用的功能。

拦截器必须实施 intercept() 方法，该方法返回 boolean 值。如果返回的值是 true，消息数据包将继续。如果返回的值是 false，则进程将被中止，不会调用其他拦截器，也不会进一步处理消息数据包。

消息拦截以透明的方式对主客户端代码发生，除非传出数据包以阻止发送模式发送。在发送传出数据包时，启用阻止并且数据包遇到返回 false 的拦截器时，会将 ActiveMQException 抛给调用者。抛出的异常包含拦截器的名称。

您的拦截器必须实现 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，或 1024 * 1024 字节。默认为大多数用例。对于其他情况，找到窗口大小的最佳值可能需要对您的系统进行基准测试。如果您需要更改默认值，AMQ 核心协议 JMS 允许您设置缓冲区窗口大小。

以下示例演示了如何使用 AMQ 核心协议 JMS 设置使用者窗口大小参数。每个示例将使用者窗口大小设置为 300,000 字节。

窗口大小根据信用额在代理和生产者之间协商，每个字节在窗口中一个得分。随着消息的发送和使用，生产者必须先向代理请求并授予其信用卡，然后才能发送更多消息。生产者和代理之间的信贷交换规范了他们之间的数据流。

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

快速消费者可以像使用它们一样快速处理消息。如果您确信消息传递系统中的使用者速度快，请考虑将窗口大小设置为 -1。将窗口大小设置为此值可在客户端上实现未绑定的消息缓冲。不过，请谨慎使用此设置。如果使用者无法在接收消息时尽快处理消息，客户端上的内存可能会溢出。

为快速使用者设置窗口大小

以下示例演示了如何使用消息快速使用者 AMQ 核心协议 JMS 客户端将窗口大小设置为 -1。

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

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

以下示例演示了如何使用 AMQ 核心协议 JMS 客户端（消息使用者较慢）将窗口大小设置为 0。

其他资源

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

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

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

以下示例显示了一个客户端，它将消息使用率限制为每秒 10 条消息。

其他资源

有关如何限制消费者速率的工作示例，请查看 <install-dir>/examples/standard 中的 consumer-rate-limit 示例。

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

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

其他资源

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