跳到主要内容

接收消息

ChatGPT-4o 中英对照 Receiving a Message

这篇文章描述了如何在 Spring 中使用 JMS 接收消息。

同步接收​

虽然 JMS 通常与异步处理相关联，但你可以同步消费消息。重载的 receive(..) 方法提供了此功能。在同步接收期间，调用线程会阻塞，直到消息变为可用。这可能是一个危险的操作，因为调用线程可能会无限期地被阻塞。receiveTimeout 属性指定接收者在放弃等待消息之前应等待多长时间。

异步接收：消息驱动的 POJO​

备注

Spring 还通过使用 @JmsListener 注解支持注解监听器端点，并提供开放的基础设施以编程方式注册端点。这是迄今为止设置异步接收器最方便的方法。有关更多详细信息，请参见启用监听器端点注解。

类似于 EJB 世界中的消息驱动 Bean（MDB），消息驱动 POJO（MDP）充当 JMS 消息的接收者。MDP 的一个限制（但请参阅 使用 MessageListenerAdapter）是它必须实现 jakarta.jms.MessageListener 接口。请注意，如果您的 POJO 在多个线程上接收消息，确保您的实现是线程安全的非常重要。

以下示例展示了一个 MDP 的简单实现：

public class ExampleListener implements MessageListener {

    public void onMessage(Message message) {
        if (message instanceof TextMessage textMessage) {
            try {
                System.out.println(textMessage.getText());
            }
            catch (JMSException ex) {
                throw new RuntimeException(ex);
            }
        }
        else {
            throw new IllegalArgumentException("Message must be of type TextMessage");
        }
    }
}
java

一旦你实现了 MessageListener，就该创建一个消息监听器容器了。

下面的示例展示了如何定义和配置 Spring 提供的一个消息监听器容器（在本例中是 DefaultMessageListenerContainer）：

@Bean
ExampleListener messageListener() {
    return new ExampleListener();
}

@Bean
DefaultMessageListenerContainer jmsContainer(ConnectionFactory connectionFactory, Destination destination,
        ExampleListener messageListener) {

    DefaultMessageListenerContainer jmsContainer = new DefaultMessageListenerContainer();
    jmsContainer.setConnectionFactory(connectionFactory);
    jmsContainer.setDestination(destination);
    jmsContainer.setMessageListener(messageListener);
    return jmsContainer;
}
java

请参阅各种消息监听器容器的 Spring javadoc（所有这些都实现了 MessageListenerContainer），以获取每个实现所支持功能的完整描述。

使用 SessionAwareMessageListener 接口​

SessionAwareMessageListener 接口是一个 Spring 特定的接口，它提供了与 JMS MessageListener 接口类似的契约，但还允许消息处理方法访问接收 Message 的 JMS Session。以下列表展示了 SessionAwareMessageListener 接口的定义：

package org.springframework.jms.listener;

public interface SessionAwareMessageListener {

    void onMessage(Message message, Session session) throws JMSException;
}
java

如果你希望你的 MDP 能够响应任何接收到的消息（通过使用 onMessage(Message, Session) 方法中提供的 Session），可以选择让你的 MDP 实现这个接口（优先于标准的 JMS MessageListener 接口）。Spring 附带的所有消息监听器容器实现都支持实现 MessageListener 或 SessionAwareMessageListener 接口的 MDP。实现 SessionAwareMessageListener 接口的类有一个警告，即它们通过接口与 Spring 绑定在一起。是否使用它完全取决于你作为应用程序开发人员或架构师的选择。

注意，SessionAwareMessageListener 接口的 onMessage(..) 方法会抛出 JMSException。与标准 JMS MessageListener 接口不同，使用 SessionAwareMessageListener 接口时，处理任何抛出的异常是客户端代码的责任。

使用 MessageListenerAdapter​

MessageListenerAdapter 类是 Spring 异步消息支持中的最终组件。简而言之，它允许你将几乎任何类公开为 MDP（尽管有一些限制）。

考虑以下接口定义：

public interface MessageDelegate {

    void handleMessage(String message);

    void handleMessage(Map message);

    void handleMessage(byte[] message);

    void handleMessage(Serializable message);
}
java

请注意，尽管该接口既没有扩展 MessageListener 也没有扩展 SessionAwareMessageListener 接口，但您仍然可以通过使用 MessageListenerAdapter 类将其用作 MDP。还要注意，各种消息处理方法是如何根据它们可以接收和处理的各种 Message 类型的内容进行强类型化的。

现在考虑以下对 MessageDelegate 接口的实现：

public class DefaultMessageDelegate implements MessageDelegate {

    @Override
    public void handleMessage(String message) {
        // ...
    }

    @Override
    public void handleMessage(Map message) {
        // ...
    }

    @Override
    public void handleMessage(byte[] message) {
        // ...
    }

    @Override
    public void handleMessage(Serializable message) {
        // ...
    }
}
java

特别注意，前面 MessageDelegate 接口（DefaultMessageDelegate 类）的实现完全没有 JMS 依赖。它确实是一个 POJO，我们可以通过以下配置将其变成一个 MDP：

@Bean
MessageListenerAdapter messageListener(DefaultMessageDelegate messageDelegate) {
    return new MessageListenerAdapter(messageDelegate);
}

@Bean
DefaultMessageListenerContainer jmsContainer(ConnectionFactory connectionFactory, Destination destination,
        ExampleListener messageListener) {

    DefaultMessageListenerContainer jmsContainer = new DefaultMessageListenerContainer();
    jmsContainer.setConnectionFactory(connectionFactory);
    jmsContainer.setDestination(destination);
    jmsContainer.setMessageListener(messageListener);
    return jmsContainer;
}
java

下一个示例展示了另一个 MDP，它只能处理接收 JMS TextMessage 消息。注意消息处理方法实际上被称为 receive（在 MessageListenerAdapter 中，消息处理方法的默认名称是 handleMessage），但它是可配置的（正如您将在本节后面看到的）。还要注意 receive(..) 方法是强类型的，只接收和响应 JMS TextMessage 消息。以下列表显示了 TextMessageDelegate 接口的定义：

public interface TextMessageDelegate {

    void receive(TextMessage message);
}
java

下面的列表显示了一个实现 TextMessageDelegate 接口的类：

public class DefaultTextMessageDelegate implements TextMessageDelegate {

    @Override
    public void receive(TextMessage message) {
        // ...
    }
}
java

随后的 MessageListenerAdapter 的配置如下：

@Bean
MessageListenerAdapter messageListener(DefaultTextMessageDelegate messageDelegate) {
    MessageListenerAdapter messageListener = new MessageListenerAdapter(messageDelegate);
    messageListener.setDefaultListenerMethod("receive");
    // We don't want automatic message context extraction
    messageListener.setMessageConverter(null);
    return messageListener;
}
java

注意，如果 messageListener 接收到的 JMS Message 类型不是 TextMessage，将抛出一个 IllegalStateException（随后被吞掉）。MessageListenerAdapter 类的另一个功能是，如果处理器方法返回一个非 void 值，则能够自动发送回一个响应 Message。请考虑以下接口和类：

public interface ResponsiveTextMessageDelegate {

    // Notice the return type...
    String receive(TextMessage message);
}
java
public class DefaultResponsiveTextMessageDelegate implements ResponsiveTextMessageDelegate {

    @Override
    public String receive(TextMessage message) {
        return "message";
    }
}
java

如果你将 DefaultResponsiveTextMessageDelegate 与 MessageListenerAdapter 一起使用，任何从 'receive(..)' 方法执行中返回的非空值（在默认配置中）都会被转换为 TextMessage。生成的 TextMessage 然后被发送到原始 Message 的 JMS Reply-To 属性中定义的 Destination（如果存在）或设置在 MessageListenerAdapter 上的默认 Destination（如果已配置）。如果未找到 Destination，则会抛出 InvalidDestinationException（请注意，此异常不会被吞掉，会在调用栈中向上传播）。

在事务中处理消息​

在事务中调用消息监听器只需要重新配置监听器容器。

您可以通过监听器容器定义中的 sessionTransacted 标志来激活本地资源事务。然后，每个消息监听器调用都在一个活动的 JMS 事务内操作，如果监听器执行失败，消息接收将回滚。发送响应消息（通过 SessionAwareMessageListener）是同一个本地事务的一部分，但任何其他资源操作（例如数据库访问）都是独立操作的。这通常需要在监听器实现中进行重复消息检测，以应对数据库处理已提交但消息处理未能提交的情况。

考虑以下 bean 定义：

@Bean
DefaultMessageListenerContainer jmsContainer(ConnectionFactory connectionFactory, Destination destination,
        ExampleListener messageListener) {

    DefaultMessageListenerContainer jmsContainer = new DefaultMessageListenerContainer();
    jmsContainer.setConnectionFactory(connectionFactory);
    jmsContainer.setDestination(destination);
    jmsContainer.setMessageListener(messageListener);
    jmsContainer.setSessionTransacted(true);
    return jmsContainer;
}
java

要参与外部管理的事务，您需要配置一个事务管理器并使用支持外部管理事务的侦听器容器（通常是 DefaultMessageListenerContainer）。

要配置一个消息监听器容器以参与 XA 事务，您需要配置一个 JtaTransactionManager（默认情况下，它会委托给 Jakarta EE 服务器的事务子系统）。请注意，底层的 JMS ConnectionFactory 需要具备 XA 能力，并且正确注册到您的 JTA 事务协调器中。（请检查您的 Jakarta EE 服务器的 JNDI 资源配置。）这使得消息接收以及（例如）数据库访问可以成为同一事务的一部分（具有统一的提交语义，但以 XA 事务日志开销为代价）。

以下 bean 定义创建了一个事务管理器：

@Bean
JtaTransactionManager transactionManager()  {
    return new JtaTransactionManager();
}
java

然后我们需要将其添加到我们之前的容器配置中。容器会处理其余的工作。以下示例展示了如何操作：

@Bean
DefaultMessageListenerContainer jmsContainer(ConnectionFactory connectionFactory, Destination destination,
        ExampleListener messageListener) {

    DefaultMessageListenerContainer jmsContainer = new DefaultMessageListenerContainer();
    jmsContainer.setConnectionFactory(connectionFactory);
    jmsContainer.setDestination(destination);
    jmsContainer.setMessageListener(messageListener);
    jmsContainer.setSessionTransacted(true);
    return jmsContainer;
}
java