0

我正在使用Alpakka它的JMS连接器来将数据从Oracle AQ. 我可以按照本指南提出以下非常基本的实现。

我的问题是如何使它具有事务性，这样我就可以保证如果抛出异常我的消息不会丢失。

object ConsumerApp extends App {
    implicit val system: ActorSystem = ActorSystem("actor-system")
    implicit val materializer: ActorMaterializer = ActorMaterializer()

    val connectionFactory = AQjmsFactory.getConnectionFactory(getOracleDataSource())

    val out = JmsSource.textSource(
        JmsSourceSettings(connectionFactory).withQueue("My_Queue")
    )

    val sink = Sink.foreach { message: String =>
        println("in sink: " + message)
        throw new Exception("") // !!! MESSAGE IS LOST !!!
    }

    out.runWith(sink, materializer)
}

如果是PL/SQL，解决方案将是这样的：

DECLARE
  dequeue_options            DBMS_AQ.DEQUEUE_OPTIONS_T;
  message_properties         DBMS_AQ.MESSAGE_PROPERTIES_T;
  message_handle             RAW (44);
  msg                        SYS.AQ$_JMS_TEXT_MESSAGE;
BEGIN
  DBMS_AQ.dequeue (
      queue_name           => 'My_Queue',
      dequeue_options      => dequeue_options,
      message_properties   => message_properties,
      payload              => msg,
      msgid                => message_handle
  );

  -- do something with the message

  COMMIT;
END;
4

1 回答 1

2

流阶段失败时的默认行为是关闭整个流。您必须决定如何处理流中的错误。例如，一种方法是使用退避策略重新启动流。

此外，由于您使用的是 Alpakka JMS 连接器，请将确认模式设置为ClientAcknowledge（自 Alpakka 0.15 起可用）。使用此配置，未确认的消息可以通过 JMS 源再次传递。例如：

val jmsSource: Source[Message, NotUsed] = JmsSource(
  JmsSourceSettings(connectionFactory)
    .withQueue("My_Queue")
    .withAcknowledgeMode(AcknowledgeMode.ClientAcknowledge)
)

val result = jmsSource
  .map {
    case textMessage: TextMessage =>
      val text = textMessage.getText
      textMessage.acknowledge()
      text
  }
  .runForeach(println)
于 2017-12-06T22:18:49.130 回答