如何调查 Apache Camel Route 上的数据？

Question

我的项目正在将数据从一个系统传输到另一个系统。我们使用 Apache Camel Routes 在 JBoss EAP v7 服务器之间发送数据。我的问题是，有没有办法调查包裹在不同路线时的内容？

我们已尝试增加日志记录，但我们的文件/控制台刚刚被淹没。我们还尝试在服务器上使用 Hawtio 来查看经过路由的消息，但未能成功识别我们的消息“卡住”的位置。

感谢任何帮助！

Answer 1

您可以使用单元测试在本地测试您的路由，然后使用adviceWith 和weave 方法在特定点记录交换的内容。

借助单元测试，即使您在 Karaf 或 Red Hat fuse 之类的东西中运行骆驼，您也可以轻松地在您最喜欢的 IDE 中调试您的路线。

package com.example;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.RoutesBuilder;
import org.apache.camel.builder.AdviceWithRouteBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.model.dataformat.JsonLibrary;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.junit.Test;

public class ExampleRouteTests extends CamelTestSupport {
    
    @Test
    public void exampleTest() throws Exception 
    {
        ContractDetails testDetails = new ContractDetails(1512, 1215);
        mockJDBCEndpoints();

        context.getRouteDefinition("exampleRoute")
            .adviceWith(context, new AdviceWithRouteBuilder(){

            @Override
            public void configure() throws Exception {
                
                replaceFromWith("direct:start");

                weaveByToUri("direct:getDetailsFromAPI")
                    .replace()
                    .to("log:testLogger?showAll=true")
                    .to("mock:api")
                    .setBody(constant(testDetails));

                weaveByToUri("direct:saveToDatabase")
                    .replace()
                    .to("log:testLogger?showAll=true")
                    .to("mock:db");
            }
        });

        MockEndpoint apiMockEndpoint = getMockEndpoint("mock:api");
        apiMockEndpoint.expectedMessageCount(1);

        MockEndpoint dbMockEndpoint = getMockEndpoint("mock:db");
        dbMockEndpoint.expectedMessageCount(1);
        
        context.start();

        String body = "{\"name\":\"Bob\",\"age\":10}";
        template.sendBody("direct:start", body);

        apiMockEndpoint.assertIsSatisfied();
        dbMockEndpoint.assertIsSatisfied();
    }  

    @Override
    protected RoutesBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder(){

            @Override
            public void configure() throws Exception {
                
                from("amqp:queue:example")
                    .routeId("exampleRoute")
                    .unmarshal().json(JsonLibrary.Jackson, 
                        Person.class)
                    .to("direct:getDetailsFromAPI")
                    .process(new SomeProcessor())
                    .to("direct:saveToDatabase");
                    
                from("direct:saveToDatabase")
                    .routeId("saveToDatabaseRoute")
                    .to("velocity:sql/insertQueryTemplate.vt")
                    .to("jdbc:exampleDatabase");

                from("direct:getDetailsFromAPI")
                    .removeHeaders("*")
                    .toD("http4:someAPI?name=${body.getName()}")
                    .unmarshal().json(JsonLibrary.Jackson, 
                        ContractDetails.class);
            }
        };
    }

    void mockJDBCEndpoints() throws Exception {

        context.getRouteDefinition("saveToDatabaseRoute")
            .adviceWith(context, new AdviceWithRouteBuilder(){

                @Override
                public void configure() throws Exception {
                    weaveByToUri("jdbc:*")
                        .replace()
                        .to("mock:db");
                }
        });
    }

    @Override
    public boolean isUseAdviceWith() {
        return true;
    }
}

现在，对于单元测试未发生的问题进行故障排除，您可以使用onException 配置通用或路由特定异常处理，并使用Dead letter channel 来处理和存储有关失败交换的信息。或者，您可以只使用stream 或文件组件将有关异常和交换失败的信息保存到单独的文件中，以避免日志泛滥。