标签: apache-camel

我对嵌入式Camel路由中的远程JMS队列有持久的消费者.是否可以使用主从配置进行此类路由？现在看来,当启动从属ActiveMQ时,Camel路由已经启动并激活,而不是在发生实际故障转移时.

现在它导致从属实例接收也发送给主服务器的相同消息,这会导致重复消息在故障转移时到达队列.

我正在使用ActiveMQ 5.3和Apache Camel 2.1.

作为Camel和ServiceMix的用户,我经常需要监控这些工具以用于开发目的.我主要使用JConsole从ServiceMix和Camel访问MBean.它为ESB开发人员提供了一些非常有用的指标.

我还使用karaf(web)控制台来跟踪我的路线和捆绑状态.环顾StackOverflow,我发现了一个有趣的链接,解释了如何为ServiceMix 4设置Hyperic HQ:使用Hyperic HQ监控Servicemix.

它们是您认为值得学习/用于Camel/SMX开发的其他工具吗？

我有一个REST服务器,它在响应正文中发送JSON.我最近开始阅读Apache Camel.我使用以下命令将请求发送到我的REST服务.

from("direct:start").setHeader("token", simple("234da"))
                            .to("http://localhost:8088/foo/bar/?foo1=bar1");

现在响应将是一个JSON,有没有什么方法我可以使用某种方法直接将这个JSON直接放入POJO to()(类似这样的东西)？

to("http://localhost:8088/foo/bar/?foo1=bar1").toPOJO();

我更喜欢非Spring解决方案.

谢谢

我的问题非常类似于http://camel.465427.n5.nabble.com/FTP-ConsumerTemplate-Threads-remaining-alive-td5548338.html.

你如何杀死不同端点上的Camel锁定线程？我尝试删除端点并在设置doneUoW后停止ConsumerTemplate,但这些都不起作用.

我们目前正在运行由一组JMS队列(无主题)连接的分布式Java应用程序.Java应用程序大量使用Apache Camel框架,该框架也用于通过JMS进行通信.作为JMS提供商,我们正在使用ActiveMQ.我们对排队解决方案的要求在功能方面非常基础 - 在某种程度上可与本地Java队列和Camel SEDA队列相媲美:

1. 队列需要高度可用
2. 平均而言,消息非常小(几KB),但不得对大小进行限制(即,传输的消息的一小部分有几个MB)
3. 交付必须"恰好一次"(例如,亚马逊SQS,仅保证"至少一次",这对我们来说是不可接受的)

我们不需要持久队列,因为消息在进入排队系统之前存储在数据库中.到目前为止,ActiveMQ在设置上花了一些时间后对我们来说效果很好.但是,有两个问题因为我们正在寻找不同的解决方案:

• 我们目前通过在JDBC Master-Slave-Mode中运行AMQ来实现高可用性.故障转移工作正常,但不如所希望的那样透明.
• JDBC存储对性能有负面影响,特别是如果我们在几个(> 6)队列上传输更大的消息(> = 2-3 MB).此外,我们有不必要地打扰我们的数据库的感觉.

我们知道AMQ从版本5.9开始支持使用Zookeeper的HA.不过,不久前,我们开始考虑/评估Hazelcast作为AMQ的替代品.很长一段时间,Hazelcast看起来非常迷人,因为它似乎满足了我们所有的要求,并且似乎比使用Zookeeper的AMQ更简单.此外,Camel通过其Hazelcast SEDA功能为使用Hazelcast作为排队解决方案提供了很好的支持.鉴于Hazelcast似乎是我们完美替代AMQ的事实,我们不知道网上没有太多可用的信息讨论Hazelcast作为JMS的(严重)替代品.因此,我们想知道:

• 用Hazelcast替换JMS有什么缺点(性能,可靠性......)？
• 是否有人在Hazelcast中遇到过多MB消息的问题？
• 有没有人遇到过Hazelcast的任何可靠性问题？我们目前仍在使用2.6(因为Camel组件当前正在使用该版本)并且已经在Ubuntu上进行了守护.到目前为止它似乎运行得很完美,但是一些额外的经验会很好听.

我有2个POJO:

public class Witch {
    private Broom broom;
    private List<Spell> spells;

    // constructors, getters/setters, etc.
}

public class ValidatedWitches {
    private List<Witch> goodWitches
    private List<Witch> badWitches;

    // constructors, getters/setters, etc.
}

我有一个我写的Camel处理器,它将生成一个ValidatedWitches实例(同样由2组成List<Witch>):

public class WitchValidator implements Processor {
    @Override
    public void process(Exchange exchange) {
        List<Witch> witchesToValidate = (List<Witch>)exchange.getIn().getBody();

        ValidatedWitches validated = validate(witchesToValidate);

        exchange.getOut().setBody(validated);
    }

    private ValidatedWitches validate(List<Witch> toValidate) {
        // For each witch, determines if it is a good witch, or a bad witch,
        // and …

Apache Camel 中是否有任何内置机制可以在上次路由处理期间发生异常时延迟路由的下一次触发时间？

例如，如果我有一条路线：

• 每 5 秒触发一次
• 从数据库中获取数据
• 处理数据并构建消息
• 向远程 ActiveMQ 实例发送消息

我希望能够处理这种情况：

如果数据库或ActiveMQ会宕机（维护或一些网络问题），导致路由处理过程中抛出异常，我想暂时延迟下次触发路由（例如等待5分钟再触发，给一些时间重新启动数据库\activemq) 并切换回 5 秒触发，当路由再次处理无异常时。或者甚至在每次失败的处理尝试中逐渐增加下一次触发的等待时间，没有失败（1 分钟/5 分钟/10 分钟），并在一切正常时切换回 5 秒的默认值。

我可以使用处理器、onException处理程序、一些数组来构建我自己的解决方法，以记住每个 routeId 的当前延迟状态以及choice()如果延迟则跳过整个路由运行的谓词，但也许Apache Camel 中有一些内置机制，因为它听起来像通常的用例场景？

我已经安装了具有单主集群配置的普通 Kubernetes。将 kamel 二进制文件存储到“/usr/local/bin”后，我可以运行 kamel 命令。我执行了“kamel install --cluster-setup”，设置很好。然后，当我尝试运行命令“kamel install”时，出现以下错误。

root@camelk:~# kamel install
Error: cannot find automatically a registry where to push images
Usage:
  kamel install [flags]

Flags:
      --base-image string          Set the base image used to run integrations
      --build-strategy string      Set the build strategy
      --build-timeout string       Set how long the build process can last
      --camel-version string       Set the camel version
      --cluster-setup              Execute cluster-wide operations only (may require admin rights)
      --context strings            Add a camel context to build at startup, by default …

我想在高可用性（HA）模式下构建应用程序。这意味着我可以拥有可变数量的实例，即我需要有 5 个应用程序实例。应用程序应从 ftp/sftp 读取数据，避免重复（一个文件不能被处理两次）。为了解决这个问题，我决定在设置中使用集群骆驼路由active/active。此设置使用幂等存储库。

下面是我的幂等存储库配置（我使用的是 spring boot、camel spring boot starters、sql 和 ftp）

@Configuration
public class IdempotentRepoConf {

    @Autowired
    private DataSource dataSource;

    @Bean
    public JdbcMessageIdRepository sftpProcesorName() {
        return new JdbcMessageIdRepository(dataSource, "sftpProcesorName");
    }

}

还有我的路由器

@Component
public class FooSftpRouter extends RouteBuilder {

    @Autowired
    private IdempotentRepository idempotentRepository;

    @Override
    public void configure() throws Exception {
        from("sftp:localhost:2221/upload/files/foo?username=foo" +
                "&password=pass" +
                "&move=./.done" +
                "&moveFailed=.error" +
                "&idempotentRepository=#sftpProcesorName")
                .idempotentConsumer(header(Exchange.FILE_NAME),idempotentRepository)
                    .to("sftp:localhost:2221/upload/files/bar?username=foo&password=pass")
                .end();

    }
}

当我只运行一个实例时，一切正常，没有任何问题，但是当我运行多个实例时，我会遇到错误

2020-08-03 15:14:14.589  WARN 18071 --- [pload/files/foo] o.a.c.c.file.remote.SftpConsumer         : Error processing file …

SpringBoot + ApacheCamel 我对 *CamelContext 有问题，但出现异常：

Cannot cast org.apache.camel.spring.SpringCamelContext to org.apache.camel.ExtendedCamelContext

在初始化一些组件时。我的配置部分（pom.xml）：

  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.3.2.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
  </parent>

...

<dependency>
  <groupId>org.apache.camel</groupId>
  <artifactId>camel-spring-boot-starter</artifactId>
  <version>2.25.2</version>
</dependency>

...

问题：如何使用这些上下文/应该使用哪一个？

• CamelContext又名DefaultCamelContext
• ExtendedCamelContext
• SpringCamelContext