java - 成功处理 Camel 内部重新传递后如何防止来自 MQ Broker 的消息重新传递(事务性 Camel 路由)

Question

我正在尝试使用JPATransactionManager运行camel transacted()路由(一个独立的java进程)，JPATransactionManager是spring PlatformTransactionManager(因为我希望camel路由在单个数据库事务中运行)，但我无法在事务方法失败的情况下抑制MQ Broker的重新传递，即使我在onException子句中使用了handled(true)以及我的自定义重新传递策略(已成功执行)。我只希望 MQ 在服务崩溃时重新交付。

尝试了以下方法，但不起作用:

• 在JMSComponent配置中设置setTransacted(false)以防止camel jms运行是transacted_session jms模式，但不起作用
• doTry 和 doCatch 事务 block 中的异常

• Camel 重新交付，然后处理(true)。

  onException(Exception.class)
      .log("ERROR OCCURRED")
      .redeliveryPolicyRef("myRedeliveryPolicy")
      .handled(true)
      .to(getPostExceptionRoute());
  
  @Bean
  @Autowired
  public RedeliveryPolicy myRedeliveryPolicy() {
      RedeliveryPolicy myRedeliveryPolicy= new RedeliveryPolicy();
      myRedeliveryPolicy.setMaximumRedeliveries(2);
      myRedeliveryPolicy.setMaximumRedeliveryDelay(2000);
      return myRedeliveryPolicy;
  }
  
  @Bean
  @Autowired
  public JmsComponent jms(IJMSConnectionFactory cf) throws JMSException {
      JmsComponent jmsComponent = new JmsComponent();
      jmsComponent.setConfiguration(jmsConfig(cf));
      jmsComponent.setTransacted(false);
      return jmsComponent;
  }
  
  from("jms:queue:TestQueue?acknowledgementModeName=CLIENT_ACKNOWLEDGE")
      .unmarshal().json(JsonLibrary.Jackson, TestObject.class)
      .transacted()
      .processRef("myPersistInDBProcessor")
  

• 我希望 Camel 根据重新交付策略尝试重新交付(有效)，但 MQ 不应重新交付。

• 我希望我的 Camel 路由在单个数据库事务中运行。
• 我希望 MQ 代理仅在我的 java 服务在处理过程中崩溃时重新传送，这样我就不会丢失消息。

Answer 1

I expect camel to try redeliveries as per redelivery policy (working) but MQ should not redeliver

当 MQ 绝不能执行重新传递(因为您在 Camel 中处理错误)时，您应该删除 acknowledgementModeName=CLIENT_ACKNOWLEDGE 或显式设置 AUTO_ACKNOWLEDGE(默认值)。

只要消息没有被确认，从代理的角度来看，它就没有被传递。 AUTO_ACKNOWLEDGE 在消费后立即确认消息，如果您从不想要重新投递，这是有意义的。

另一方面，

CLIENT_ACKNOWLEDGE 仅在特定条件下确认消息，请参阅 this post for some more info about this .

由于评论新信息而进行编辑

如果您想要 MQ 重新传递，但在大多数情况下使用 Camel“覆盖”它们，则必须使用已处理的消息。

通过像这样配置 JMS 组件来使用本地 JMS 代理事务

jmsComponent.setLazyCreateTransactionManager(false);
jmsComponent.setTransacted(true);

对于这种类型的事务，您根本不需要 Spring TransactionManager。所以我猜想 JPATransactionManager 被 JMS 忽略了，并且您的 JMS 消耗应该是事务性的。

现在，当您的 Camel 错误处理程序通过使用 handled(true)“吞掉”异常时，一定不会有 MQ 重新传递。但是当异常传播回代理时，MQ 会重新传递。

I expect my camel route to run in a single db transaction

我在您的问题中没有找到任何有关数据库事务不起作用的内容。似乎只有一个处理器负责数据库工作。如果此方法无法正常工作，请在您的问题或单独的问题中描述问题。