第 6 章 Camel 组件

@Startup
@ApplicationScoped
@ContextName("activemq-camel-context")
public class ActiveMQRouteBuilder extends RouteBuilder {

  @Override
  public void configure() throws Exception {
    from("timer://sendJMSMessage?fixedRate=true&period=10000")
    .transform(constant("<?xml version='1.0><message><greeting>hello world</greeting></message>"))
    .to("activemq:queue:WildFlyCamelQueue?brokerURL=vm://localhost")
    .log("JMS Message sent");
  }
}

每次将消息添加到 WildFlyCamelQueue 目的地时，都会向控制台输出一条日志消息。要验证消息实际上被放入队列中，您可以使用 Camel on EAP 子系统提供的 ../features/hawtio.md[Hawtio 控制台,window=_blank]。

要使用 ActiveMQ 消息，Camel RouteBuilder 实施与制作者示例类似。

当 ActiveMQ 端点消耗来自 WildFlyCamelQueue 目的地的消息时，内容会记录到控制台。

@Override
public void configure() throws Exception {
  from("activemq:queue:WildFlyCamelQueue?brokerURL=vm://localhost")
  .to("log:jms?showAll=true");
}

camel-activemq 组件需要类型为 org.springframework.transaction.PlatformTransactionManager 的事务管理器。因此，您可以先创建一个可满足此要求的 bean 扩展 JtaTransactionManager。请注意，bean 被标注为 @Named，以允许在 Camel bean registry 中注册 bean。另请注意，JBoss EAP 事务管理器和用户事务实例使用 CDI 注入。

@Named("transactionManager")
public class CdiTransactionManager extends JtaTransactionManager {

  @Resource(mappedName = "java:/TransactionManager")
  private TransactionManager transactionManager;

  @Resource
  private UserTransaction userTransaction;

  @PostConstruct
  public void initTransactionManager() {
    setTransactionManager(transactionManager);
    setUserTransaction(userTransaction);
  }
}

接下来，您需要声明要使用的事务策略。同样，使用 @Named 注释使 bean 可用于 Camel。事务管理器也会注入，以便使用所需的事务策略创建 TransactionTemplate。此实例中 PROPAGATION_REQUIRED。

@Named("PROPAGATION_REQUIRED")
public class CdiRequiredPolicy extends SpringTransactionPolicy {
  @Inject
  public CdiRequiredPolicy(CdiTransactionManager cdiTransactionManager) {
    super(new TransactionTemplate(cdiTransactionManager,
      new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRED)));
  }
}

现在，您可以配置 Camel RouteBuilder 类，并注入 Camel ActiveMQ 组件所需的依赖项。在资源适配器配置上配置的 ActiveMQ 连接工厂会与您之前配置的事务管理器一起注入。

在本例中，当任何消息都从 queue1 使用时，它们都会路由到另一个名为 queue2 的 JMS 队列。从 queue2 使用的消息会导致 JMS 事务通过 rollback （）DSL 方法回滚。这将使原始消息放置在死信队列(DLQ)上。

@Startup
@ApplicationScoped
@ContextName("activemq-camel-context")
public class ActiveMQRouteBuilder extends RouteBuilder {

  @Resource(mappedName = "java:/ActiveMQConnectionFactory")
  private ConnectionFactory connectionFactory;

  @Inject
  private CdiTransactionManager transactionManager;

  @Override
  public void configure() throws Exception {
    ActiveMQComponent activeMQComponent = ActiveMQComponent.activeMQComponent();
    activeMQComponent.setTransacted(false);
    activeMQComponent.setConnectionFactory(connectionFactory);
    activeMQComponent.setTransactionManager(transactionManager);

    getContext().addComponent("activemq", activeMQComponent);

      errorHandler(deadLetterChannel("activemq:queue:ActiveMQ.DLQ")
      .useOriginalMessage()
      .maximumRedeliveries(0)
      .redeliveryDelay(1000));

    from("activemq:queue:queue1F")
      .transacted("PROPAGATION_REQUIRED")
      .to("activemq:queue:queue2");

    from("activemq:queue:queue2")
      .to("log:end")
      .rollback();
  }
}