Apache Camel路由Mock测试空指针异常问题排查

问题描述 #

我编写了包含SEDA端点、多播、线程池、基于内容路由的Apache Camel路由，其中：

• SEDA_SEND_ENDPOINT：seda:sendMessage?blockWhenFull=true&concurrentConsumers=100
• SEDA_PROCESS_ENDPOINT：seda:processMessage?blockWhenFull=true

尝试编写Mock测试用例模拟seda:processMessage端点时，运行测试出现空指针异常，怀疑测试未正确加载路由，需确认测试代码调用是否正确。

路由代码 #

// SEDA Endpoint Stage Event Driven Architecture
from(Endpoints.SEDA_SEND_ENDPOINT)
    .messageHistory()
    // 路由名称
    .routeId(Endpoints.SEDA_SEND_ENDPOINT)
    .log("${body}")

    // 多播
    .multicast()
    .parallelProcessing() // 创建并行线程
    .log("${body}")

    // 线程池
    .threads()
    .executorService(executorService) // 指定线程池
    .log("Camel Route Started Message Processing : - ${body}")

    // 基于内容的路由
    .choice()
    .when(CommonUtility.costIQPredicate) // 根据header值判断路由的断言
    // .bean(CostIQService.class, "calculatePrice") // 规则引擎调用
    .bean(CostIQPayloadTransformer.class, "payloadTransformer") // 载荷转换

    // 多消费者
    .to(
        Endpoints.SEDA_PROCESS_ENDPOINT, // 消费者1
        Endpoints.SEDA_PROCESS_ENDPOINT, // 消费者2
        Endpoints.SEDA_PROCESS_ENDPOINT) // 消费者3
    .when(CommonUtility.optimizationPredicate)
    .bean(OptimizationService.class, "calculatePrice")
    .bean(CostIQPayloadTransformer.class, "payloadTransformer")
    .to(
        Endpoints.SEDA_PROCESS_ENDPOINT,
        Endpoints.SEDA_PROCESS_ENDPOINT,
        Endpoints.SEDA_PROCESS_ENDPOINT)
    .when(CommonUtility.markDownPredicate)
    .bean(MarkDownService.class, "calculatePrice")
    .bean(CostIQPayloadTransformer.class, "payloadTransformer")
    .to(
        Endpoints.SEDA_PROCESS_ENDPOINT,
        Endpoints.SEDA_PROCESS_ENDPOINT,
        Endpoints.SEDA_PROCESS_ENDPOINT)
    .when(CommonUtility.pricingPredicate)
    .bean(PricingService.class, "calculatePrice")
    .bean(CostIQPayloadTransformer.class, "payloadTransformer")
    .to(
        Endpoints.SEDA_PROCESS_ENDPOINT,
        Endpoints.SEDA_PROCESS_ENDPOINT,
        Endpoints.SEDA_PROCESS_ENDPOINT)
    .log("Final :- ${body}")
    .end();

原测试代码 #

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.HashMap;
import java.util.Map;

import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.junit5.CamelTestSupport;
import org.apache.camel.test.spring.junit5.MockEndpoints;
import org.junit.jupiter.api.Test;

import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.Predicate;
import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;

//@SpringBootTest
// @RunWith(CamelSpringBootRunner.class)
@MockEndpoints("seda:processMessage")
public class CamelRouteTests1 extends CamelTestSupport {

        @EndpointInject(uri = "mock:seda:processMessage")
        MockEndpoint mock;
    
        @Autowired
        ProducerTemplate template;
    
        @Test
        public void testMethod() throws InterruptedException {
          mock.expectedBodiesReceived("test");
       
          template.sendBody("seda:sendMessage", "test");
    
          mock.assertIsSatisfied();
        }
}

问题分析 #

空指针异常的核心原因是测试代码混淆了纯Camel测试框架（CamelTestSupport）和Spring测试注解的使用：

1. CamelTestSupport不会自动加载Spring上下文，因此@Autowired注入的ProducerTemplate会为null，触发空指针。
2. 测试类未加载目标路由，Camel上下文为空，消息无法进入路由处理。
3. @MockEndpoints是Spring集成测试注解，在纯CamelTestSupport环境下无法生效，mock对象同样为null。

修复方案 #

根据场景推荐两种测试方式，选其一即可：

方式一：使用CamelTestSupport加载路由并替换端点 #

手动添加路由定义，通过adviceWith将seda:processMessage替换为mock端点：

import org.apache.camel.CamelContext;
import org.apache.camel.builder.AdviceWith;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.junit5.CamelTestSupport;
import org.junit.jupiter.api.Test;

public class CamelRouteTests1 extends CamelTestSupport {

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        // 加载你的实际路由类，假设路由在MyCamelRoute类中
        return new MyCamelRoute();
    }

    @Test
    public void testMethod() throws InterruptedException {
        CamelContext context = context();
        // 替换seda:processMessage为mock端点
        AdviceWith.adviceWith(context, Endpoints.SEDA_SEND_ENDPOINT, route -> {
            route.weaveByToUri("seda:processMessage*")
                 .replace()
                 .to("mock:processMessage");
        });

        MockEndpoint mock = getMockEndpoint("mock:processMessage");
        // 匹配一个predicate会发送3次消息到processMessage，设置预期数量
        mock.expectedMessageCount(3);
        // 若payloadTransformer不修改body，直接预期原内容
        mock.expectedBodiesReceived("test");

        // 发送测试消息
        template.sendBody("seda:sendMessage", "test");

        mock.assertIsSatisfied();
    }
}

方式二：使用Spring Boot集成测试 #

如果是Spring Boot项目，改用Spring Boot测试框架确保上下文正确加载：

import org.apache.camel.ProducerTemplate;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
import org.apache.camel.test.spring.junit5.MockEndpoints;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;

// 启用Spring Boot测试，加载应用上下文
@CamelSpringBootTest
// 自动将seda:processMessage替换为mock端点
@MockEndpoints("seda:processMessage")
public class CamelRouteTests1 {

    @Autowired
    private MockEndpoint mockSedaProcessMessage;

    @Autowired
    private ProducerTemplate producerTemplate;

    @Test
    public void testMethod() throws InterruptedException {
        // 匹配一个predicate会发送3次消息，设置预期数量
        mockSedaProcessMessage.expectedMessageCount(3);
        mockSedaProcessMessage.expectedBodiesReceived("test");

        producerTemplate.sendBody("seda:sendMessage", "test");

        mockSedaProcessMessage.assertIsSatisfied();
    }
}

额外注意事项 #

1. 确保测试消息的header值能匹配某个定义的predicate（如costIQPredicate），否则消息会走到choice默认分支，不会发送到seda:processMessage。可添加对应header：

producerTemplate.sendBodyAndHeader("seda:sendMessage", "test", "yourHeaderKey", "valueThatMatchesPredicate");

2. 若CostIQPayloadTransformer修改消息体，需调整expectedBodiesReceived的内容为转换后的结果。
3. 多播+并行线程为异步处理，可设置超时时间确保测试等待完成：mock.assertIsSatisfied(10000);（10秒超时）。

内容的提问来源于stack exchange，提问作者John