标签: apache-camel

我一直在从事一个项目，该项目从各种第三方数据源收集数据并将数据挖掘到我们的数据存储 (DI) 中。为此，我们一直在使用 Pentaho。

我想知道这是否也可以用 ESB（骆驼或骡子）来完成？ESB 带来了哪些 DI 没有提供的其他特性？

我已经阅读了很多关于 ESB 和 DI 的文章，但没有一篇能够解决这个查询。我还阅读了有关第三方数据源的 mule 数据连接器的信息。

我正在评估一个基于 EIP（企业集成模式）用于我的应用程序的集成框架。我有 apache camel 和 spring 集成作为候选框架。

我看到 spring 集成提供了或多或少类似于 EIP 的功能，如消息、频道等。

但是，apache camel 没有公开任何功能，如 message 、 channel 、 endpoints 等。相反，它似乎以自己的方式使概念变得更简单。

我觉得 spring 集成更像是 EIP 的真正实现，而不是骆驼。

不确定与骆驼相比，弹簧集成是否有缺点。

任何人都可以帮助我决定使用它们中的每一个都比另一个更合适的用例或场景。

此外，我的另一个查询确实使用它们中的任何一个作为抽象，如果以后我考虑使用像rabbitmq或activemq这样的消息传递系统，我认为它们也具有类似的功能。

谢谢，斯里

我有一条骆驼路线，如下所示

from("activemq:queue:upload" )
    .pollEnrich().simple("file:basePath/${header.ID}?noop=true&recursive=true")
    .aggregationStrategy(new ExampleAggregationStrategy()) 
    .timeout(2000) 
 .toD("ftp:${header.destinationURI}")

在我的文件系统中file:basePath/${header.ID}包含多个文件夹。执行上述路由时，只会将第一个文件夹中的第一个文件复制到 ftp 服务器。剩余的文件夹（带子文件夹）没有被复制到 ftp 服务器！

而ExampleAggregationStrategy()类的aggregate()方法看起来像以下

@Override
        public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
            String destinationURI = "populatedURI";

        oldExchange.setOut(newExchange.getIn());
        oldExchange.getOut().setHeader("ID",  oldExchange.getIn().getHeader("ID"));
        oldExchange.getOut().setHeader("destinationURI", destinationURI);
        oldExchange.setFromEndpoint(newExchange.getFromEndpoint());
        oldExchange.setFromRouteId(newExchange.getFromRouteId());

            return oldExchange;
        }

我也试过设置properties and onCompletions。仍然没有运气！我错过了什么aggregationStrategy吗？
如何成功复制所有文件和文件夹pollEnrich？

我有一个骆驼路线，它有一个步骤调用子路线将正文的文本部分转换为 PDF。不幸的camel-pdf是不保留标题。有没有办法在不丢失当前交换的情况下获得子路由的价值？

子路由

from("seda:generate-pdf")

// Back up the original in a header
.setHeader("original", body())

// Create the PDF
.to("pdf:create?textProcessingFactory=autoFormatting")
// UHOH! All my headers are gone :(

// Set the PDF as the header for the doc server
.setHeader("pdf", body())

// Move the indicator back to the body
.setBody(header("original")) // <-- this no longer exists

主要路线

// Snip

// Unmarshal into Java

.unmarshal().json(JsonLibrary.Gson, MyReportContainingText.class)

// Call sub-route to generate the PDF
.inOut("seda:generate-pdf")
// UHOH! All my headers are …

我正在与使用骆驼的 Talend ESB 合作。我的要求是我需要对表进行批量插入，并且必须为不同类型的输入动态创建查询。

如果我创建一个上下文说 extVar 并将其默认为“插入 table_foo (foo, bar) 值 (:#foo, :#bar)”，然后使用带有以下代码的 cMessagingEndpoint 组件“sql”+context.extVar+”?batch= true&dataSource=mysql”它工作正常。

在 Talend 中，我能够访问 cProcessor 内的 context.extVar，因此我尝试从那里动态加载查询，但我不能。

由于我使用的是命名参数，因此我假设我既不能使用主体进行查询，也不能在 cMessageEndPoint 的 sql 语句中获取属性/标头值。

有什么方法可以将查询动态填充到 cMessageEndPoint 的 sql 语句中。

我正在尝试使用http://camel.apache.org/mock.html为我的骆驼路线创建测试用例。我需要验证路由中的处理器。但是简单的测试对我不起作用。

public class CamelRouteTest  extends CamelTestSupport {

  @Override
  public String isMockEndpointsAndSkip() {
    // override this method and return the pattern for which endpoints to mock,
    // and skip sending to the original endpoint.
    return "mock:result";
  }

  @Test
  public void verifyMessageCount() throws Exception {
    template.sendBody("Test");
    getMockEndpoint("mock:result").expectedMessageCount(1);
    assertMockEndpointsSatisfied();
  }

  @Override
  protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start").to("mock:result");
        }
    };
  }
}

堆栈跟踪：

java.lang.IllegalArgumentException: defaultEndpoint must be specified
    at org.apache.camel.util.ObjectHelper.notNull(ObjectHelper.java:308) …

我必须写信给骆驼的多个 if 条件，我需要帮助才能解决问题。

if(token is NULL)
if(condition is NULL) 
if(Dates does not match)
Then execute this...

我正在尝试的是

.choice
.when(token is NULL)
.when(condition is NULL)
.when(Dates does not match)
.log(update DB)
.endchoice()

哪个不起作用..请帮忙

我有一个 Springboot 应用程序，其中配置了一些 Camel 路由。

public class CamelConfig {
    private static final Logger LOG = LoggerFactory.getLogger(CamelConfig.class);

    @Value("${activemq.broker.url:tcp://localhost:61616}")
    String brokerUrl;

    @Value("${activemq.broker.maxconnections:1}")
    int maxConnections;

    @Bean
    ConnectionFactory jmsConnectionFactory() {
        PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(new ActiveMQConnectionFactory(brokerUrl));
        pooledConnectionFactory.setMaxConnections(maxConnections);
        return pooledConnectionFactory;
    }

    @Bean
    public RoutesBuilder route() {
        LOG.info("Initializing camel routes......................");
        return new SpringRouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("activemq:testQueue")
                  .to("bean:queueEventHandler?method=handleQueueEvent");
            }
        };
    }
}

我想测试从activemq:testQueue到 的这条路线queueEventHandler::handleQueueEvent。我尝试了这里提到的不同内容http://camel.apache.org/camel-test.html，但似乎没有让它工作。

我正在尝试做这样的事情：

    @RunWith(SpringRunner.class)
    @SpringBootTest(classes = {CamelConfig.class,   CamelTestContextBootstrapper.class})
    public class CamelRouteConfigTest {

    @Produce(uri …

根据 Camel 文档，我创建了 JaxbDataFormat（文档中的示例代码使用了不存在的构造函数？）

@Override
public void configure() throws Exception {

    JaxbDataFormat jaxbDataFormat = new JaxbDataFormat();
    jaxbDataFormat.setContextPath("somepackage");

我有 pom 依赖

    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-jaxb</artifactId>
        <version>2.18.3</version>
    </dependency>

不起作用：“ConvertBody...因为无法创建数据格式'jaxb'。”

有人可以给出一个示例代码，说明 jaxb 转换应该如何与 Camel 一起使用。我有 Camel in Action 2ed，但那里的示例使用 XML-definde 路由。使用 XML 的过程似乎很简单 - 但我对使用 xml 作为编程语言不是很热情；)

使用 Java 8。

………………

线程“CamelMainRunController”中的异常java.lang.RuntimeException：org.apache.camel.FailedToCreateRouteException：无法在以下位置创建路由route2：>>> Marshal[org.apache.camel.model.dataformat.JaxbDataFormat@57d7f108] <<< in路线：Route(route2)[[From[activemq:gateway.queue]] -> [OnException... 因为无法创建数据格式'jaxb'。确保数据格式有效并且关联的 Camel 组件存在于 org.apache.camel.spring.boot.CamelSpringBootApplicationController.run(CamelSpringBootApplicationController.java:74) 处的类路径中 org.apache.camel.spring.boot.CamelMainRunController $DaemonTask.run(CamelMainRunController.java:42) at java.lang.Thread.run(Thread.java:745) 引起：org.apache.camel.FailedToCreateRouteException：无法在以下位置创建路由 route2：>>> Marshal[org.apache.camel.model.dataformat.JaxbDataFormat@57d7f108] <<< in route: Route(route2)[[From[activemq:gateway.queue]] -> [OnException... 因为数据无法创建格式“jaxb”。确保数据格式有效，并且关联的 Camel 组件存在于 org.apache.camel.model.RouteDefinition.addRoutes(RouteDefinition.java:1071) 的类路径中，位于 org.apache.camel.model.RouteDefinition.addRoutes(RouteDefinition) .java:196) 在 org.apache.camel.impl.DefaultCamelContext.startRoute(DefaultCamelContext.java:984) …

我正在做类似的事情

  from(rabbitmq:pollingQueue?prefetchSize=1&concurrentConsumer=10)
        .process(pollingRequestStatus) // check status of the request, if not ready, requeue = true
        .Choice
           .when(requeue == true) // request not ready
           .to(rabbitmq:pollingQueue)//back to the same queue
        .endChoice
        .otherwise
        .to(proceedToSomethingElse)
        .endChoice.end;

当重新排队发生时，消息会重复，这是将消息发回同一队列时的预期行为吗？

我也按照建议尝试了以下类似的东西，但它不起作用，该消息似乎只是被消耗掉了，不会重新排队

from(rabbitmq:pollingQueue? prefetchSize=1&concurrentConsumer=10)
     .onException(NotReadyException.class)
     .handled(true)
     .setHeader(RabbitMQConstants.REQUEUE, constant(true))
     .end() 
     .process(pollingRequestStatus) // check status of the request, if not ready, throw NotReadyEception
        .to(proceedToSomethingElse);

我尝试过的另外两种方法至少不会创建重复项，

1.) 在 NotReadyExeption 上，将消息发送回 pollingQueue

from(rabbitmq:pollingQueue? prefetchSize=1&concurrentConsumer=10)
    .onException(NotReadyException.class)
    .to(rabbitmq:pollingQueue)
    //.delay(constant(8000)) //not sure why it throws error if i set delay
    .end
    .process(pollingRequestStatus); // check status …