Apache Camel动态插入数据至from后及SQL模板参数替换失效问题

首先解决你遇到的核心问题——SQL模板中:param参数替换失效的问题，然后再讲解如何在from之后动态插入数据。

一、SQL模板参数替换失效的解决方案 #

你的问题根源在于没有正确利用Camel SQL组件的参数绑定机制，或者混淆了字符串占位符与Camel参数引用的方式。下面提供两种可行方案，优先推荐安全的参数绑定方式：

方案1：使用Camel SQL组件的参数绑定（推荐，防SQL注入） #

Camel SQL组件支持通过:#headerName的语法直接引用Exchange中的header参数，同时启用useHeadersAsParameters=true来自动绑定参数。步骤如下：

1. 修正SQL模板：
   修复原模板中的语法错误（delivery_trigger_time的拼接和别名），并改用Camel SQL的参数引用格式：

   private static final String MARS_SQL_TEMPLATE_TWO = "select\n" + 
       " id_delivery_area,\n" + 
       " id_country,\n" + 
       " id_gate_sites_group,\n" + 
       " free_shipping_bill,\n" + 
       " id_territory,\n" + 
       " price,\n" + 
       " id_mtt_type,\n" + 
       " id_gt_service_level,\n" + 
       " extract (hour from delivery_trigger_time) || ':' || extract (minute from delivery_trigger_time) as delivery_trigger_time,\n" + 
       " delivery_period_to_hours,\n" + 
       " delivery_period_from_days,\n" + 
       " delivery_period_to_days,\n" + 
       " :#brand as brand,\n" + 
       " :#country as country\n" + 
       " from :#param ";
   

2. 调整路由定义：
   使用Camel SQL组件执行SQL，并开启参数绑定：

   from("direct:" + MARS_ROUTE)
       .setBody(constant(MARS_SQL_TEMPLATE_TWO))
       // 替换yourDataSourceName为你的数据源Bean名称
       .to("sql:${body}?dataSource=yourDataSourceName&useHeadersAsParameters=true")
       .convertBodyTo(List.class);
   

3. 修改调用方式：
   传递所有需要的参数（包括param、brand、country）到header中：

   // 扩展executeRoute支持多header参数
   public Object executeRoute(String endpointName, Map<String, Object> headers) {
       val template = DefaultFluentProducerTemplate.on(camelContext);
       if (headers != null) {
           headers.forEach(template::withHeader);
       }
       return template.to("direct://" + endpointName).request();
   }
   
   // 调用示例
   Map<String, Object> headers = new HashMap<>();
   headers.put("param", b); // b是你要替换的表名/子查询
   headers.put("brand", "your_brand_value");
   headers.put("country", "your_country_value");
   List<MarsDeliveryArea> finalList = (List<MarsDeliveryArea>) routeService.executeRoute(MARS_ROUTE, headers);
   

方案2：手动拼接SQL（不推荐，存在SQL注入风险） #

如果一定要使用Java的字符串占位符%s，可以在路由中通过simple语言完成拼接后再执行：

1. 保留原模板的%s占位符：
   确保模板语法正确（修复delivery_trigger_time的拼接）：

   private static final String MARS_SQL_TEMPLATE_TWO = "select\n" + 
       " id_delivery_area,\n" + 
       " id_country,\n" + 
       " id_gate_sites_group,\n" + 
       " free_shipping_bill,\n" + 
       " id_territory,\n" + 
       " price,\n" + 
       " id_mtt_type,\n" + 
       " id_gt_service_level,\n" + 
       " extract (hour from delivery_trigger_time) || ':' || extract (minute from delivery_trigger_time) as delivery_trigger_time,\n" + 
       " delivery_period_to_hours,\n" + 
       " delivery_period_from_days,\n" + 
       " delivery_period_to_days,\n" + 
       " '%s' as brand,\n" + 
       " '%s' as country\n" + 
       " from %s ";
   

2. 路由中拼接SQL：

   from("direct:" + MARS_ROUTE)
       .setBody(simple(String.format(MARS_SQL_TEMPLATE_TWO, 
           "${header.brand}", "${header.country}", "${header.param}")))
       .to("sql:${body}?dataSource=yourDataSourceName")
       .convertBodyTo(List.class);
   

3. 调用方式和方案1一致，传递对应的header参数即可。

二、在Apache Camel的from之后动态插入数据 #

在路由的from之后动态插入数据，常见的实现方式有以下几种：

1. 使用setBody/setHeader动态设置内容 #

这是最常用的方式，可以从header、exchange属性或外部来源获取数据，设置到消息体或header中：

from("direct:myRoute")
    // 从exchange属性获取数据设置到header
    .setHeader("dynamicTable", simple("${exchangeProperty.targetTable}"))
    // 动态构造SQL语句作为消息体
    .setBody(simple("SELECT * FROM ${header.dynamicTable}"))
    .to("sql:${body}");

2. 使用enrich/pollEnrich合并外部数据 #

如果需要从另一个端点获取动态数据并合并到当前消息流中，可以使用enrich（主动调用端点）或pollEnrich（主动拉取端点数据）：

from("direct:mainRoute")
    // 调用direct:getDynamicData获取数据，然后合并到原消息
    .enrich("direct:getDynamicData", (oldExchange, newExchange) -> {
        String dynamicData = newExchange.getIn().getBody(String.class);
        oldExchange.getIn().setHeader("dynamicData", dynamicData);
        return oldExchange;
    })
    .to("log:merged_message");

// 提供动态数据的子路由
from("direct:getDynamicData")
    .setBody(constant("dynamic_content_from_other_source"));

3. 自定义Processor实现复杂逻辑 #

如果需要更灵活的动态数据生成逻辑，可以编写自定义Processor：

from("direct:customRoute")
    .process(exchange -> {
        // 自定义逻辑生成动态数据
        String dynamicPayload = generateDynamicData(exchange.getIn().getHeaders());
        exchange.getIn().setBody(dynamicPayload);
    })
    .to("direct:next_processing_step");

// 自定义动态数据生成方法
private String generateDynamicData(Map<String, Object> headers) {
    // 根据header参数生成数据
    return "custom_dynamic_data_for_" + headers.get("key");
}

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