使用 Alpakka 手动确认 ActiveMQ 消息

Question

我正在 Java 中实现 Akka Alpakka 以从 ActiveMQ 队列消费和生产到 ActiveMQ 队列。可以从队列消费成功，但是还没有实现应用级的消息确认。

我的目标是使用队列中的消息并将它们发送给另一个参与者进行处理。当该参与者完成处理后，我希望它能够控制 ActiveMQ 中消息的确认。据推测，这可以通过向另一个可以进行确认的参与者发送消息、对消息本身调用确认函数或其他方式来完成。

在我的测试中，将 2 条消息放入 AlpakkaTest 队列，然后此代码尝试使用并确认它们。但是，我没有看到将 ActiveMQ 会话设置为 CLIENT_ACKNOWLEDGE 的方法，而且我没有看到调用或不调用 m.acknowledge(); 的行为有任何差异。因此，我认为消息仍在自动确认中。

有人知道为 CLIENT_ACKNOWLEDGE 配置 ActiveMQ 会话并在 Java 使用 Alpakka 的 Akka 系统中手动确认 ActiveMQ 消息的公认方法吗？

相关测试函数为：

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://0.0.0.0:2999"); // An embedded broker running in the test.

Source<Message, NotUsed> jmsSource = JmsSource.create(
    JmsSourceSettings.create(connectionFactory)
        .withQueue("AlpakkaTest")
        .withBufferSize(2)
);

Materializer materializer = ActorMaterializer.create(system); // `system` is an ActorSystem passed to the function.

try {
    List<Message> messages = jmsSource
        .take(2)
        .runWith(Sink.seq(), materializer)
        .toCompletableFuture().get(4, TimeUnit.SECONDS);

    for(Message m:messages) {
        System.out.println("Found Message ID: " + m.getJMSMessageID());

        try {
            m.acknowledge();
        } catch(JMSException jmsException) {
            System.out.println("Acknowledgement Failed for Message ID: " + m.getJMSMessageID() + " (" + jmsException.getLocalizedMessage() + ")");
        }
    }
} catch (InterruptedException e1) {
    e1.printStackTrace();
} catch (ExecutionException e1) {
    e1.printStackTrace();
} catch (TimeoutException e1) {
    e1.printStackTrace();
} catch (JMSException e) {
    e.printStackTrace();
}

此代码打印：

Found Message ID: ID:jmstest-43178-1503343061195-1:26:1:1:1
Found Message ID: ID:jmstest-43178-1503343061195-1:27:1:1:1

Answer 1

查看 Alpakka JmsSourceStage 的源代码，它已经为您确认每条传入消息（并且它的会话是客户端确认会话）。据我所知，没有任何模式允许您确认消息。

您可以查看 Alpakka 的源代码 here。

Answer 2

更新：自Alpakka 0.15以来，JMS连接器中的确认模式是可配置的。来自链接文档：

Source<Message, NotUsed> jmsSource = JmsSource.create(JmsSourceSettings
    .create(connectionFactory)
    .withQueue("test")
    .withAcknowledgeMode(AcknowledgeMode.ClientAcknowledge())
);

CompletionStage<List<String>> result = jmsSource
    .take(msgsIn.size())
    .map(message -> {
        String text = ((ActiveMQTextMessage)message).getText();
        message.acknowledge();
        return text;
    })
    .runWith(Sink.seq(), materializer);

从 0.11 版开始，Alpakka 的 JMS 连接器不支持应用程序级消息确认。 Alpakka 在内部 MessageListener 中使用 CLIENT_ACKNOWLEDGE 模式 here and acknowledges each message here 在内部创建一个 Session。 API 不会公开这些设置以供覆盖。

有一个公开的 ticket 讨论启用基于队列的源的下游确认，但该票证已经有一段时间不活动了。

目前您无法阻止 Alpakka 在 JMS 级别确认消息。但是，这并不妨碍您向流中添加一个阶段，该阶段将每条消息发送给 actor 进行处理，并将 actor 的回复用作背压信号。 Akka Streams documentation 描述了如何使用 mapAsync 和 ask 的组合或 Sink.actorRefWithAck 来做到这一点。例如，要使用前者：

Timeout askTimeout = Timeout.apply(4, TimeUnit.SECONDS);

jmsSource
    .mapAsync(2, msg -> ask(processorActor, msg, askTimeout))
    .runWith(Sink.seq(), materializer);

（旁注：在相关的 Streamz 项目中，Alpakka 文档中有一个最近打开的 ticket to allow application-level acknowledgement. Streamz is the replacement for the old akka-camel module and, like Alpakka, is built on Akka Streams. Streamz also has a Java API and is listed 作为外部连接器。）

使用 Alpakka 手动确认 ActiveMQ 消息

Manual Acknowledgement of ActiveMQ Messages with Alpakka

java

akka

activemq

akka-stream

alpakka