Apache Camel新手求助：JSON转指定分隔符CSV实现方案

嗨，作为Apache Camel新手能顺利完成JSON到POJO的反序列化已经超棒啦！我来帮你一步步搞定剩下的需求，分两部分说：先解决「无需Getter/Setter生成|分隔CSV」的核心问题，再处理自定义头/尾记录的要求。

核心思路是用Jackson的字段访问模式，它可以直接读取POJO的字段（不管有没有Getter/Setter），然后生成CSV。具体步骤如下：

1. 配置Jackson的CSV序列化规则 #

首先创建一个支持字段访问的JacksonDataFormat，指定分隔符为|：

import com.fasterxml.jackson.databind.MapperFeature;
import com.fasterxml.jackson.dataformat.csv.CsvMapper;
import com.fasterxml.jackson.dataformat.csv.CsvSchema;
import org.apache.camel.component.jackson.JacksonDataFormat;

// 初始化CSV Mapper，禁用Getter/Setter自动探测，启用字段访问
CsvMapper csvMapper = new CsvMapper();
csvMapper.configure(MapperFeature.AUTO_DETECT_GETTERS, false);
csvMapper.configure(MapperFeature.AUTO_DETECT_SETTERS, false);
csvMapper.configure(MapperFeature.AUTO_DETECT_FIELDS, true);

// 定义CSV schema，设置分隔符为|
CsvSchema csvSchema = CsvSchema.builder()
        .setColumnSeparator('|')
        .build();

// 绑定你的POJO类
JacksonDataFormat csvDataFormat = new JacksonDataFormat(csvMapper, YourEmployeePOJO.class, csvSchema);

2. 在Camel路由中使用该DataFormat #

把你已有的路由接上这个序列化逻辑就行：

from("http://localhost:8080/employeePayload")
        .unmarshal(yourExistingJsonDataFormat) // 你已经实现的JSON转POJO步骤
        .marshal(csvDataFormat) // 用上面配置的规则转成|分隔的CSV
        .to("file:/your/output/path?fileName=raw_employees.csv");

这样生成的CSV会直接用POJO的字段名作为表头（如果需要），字段值作为行内容，完全不用写Getter/Setter。

因为你需要的头是15列（和POJO的110个字段数量不同），还需要静态+动态列名，以及尾记录，所以不能用默认的CSV表头，得自己手动拼接内容。这里推荐用自定义Transformer Bean + Camel聚合策略来实现：

1. 写一个CSV转换Bean #

这个Bean负责生成头、把POJO转成15列的CSV行、生成尾记录：

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.MapperFeature;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;

public class EmployeeCsvTransformer {
    // 配置Jackson直接访问字段
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    {
        objectMapper.configure(MapperFeature.AUTO_DETECT_GETTERS, false);
        objectMapper.configure(MapperFeature.AUTO_DETECT_SETTERS, false);
        objectMapper.configure(MapperFeature.AUTO_DETECT_FIELDS, true);
    }

    // 生成自定义头记录（15列，静态+动态示例）
    public String generateHeader() {
        // 静态列名+动态列（比如当前日期）
        return "EMP_ID|NAME|DEPT|EMAIL|PHONE|HIRE_DATE|SALARY|MANAGER_ID|LOCATION|JOB_TITLE|EMP_TYPE|JOIN_YEAR|STATUS|COUNTRY|GENERATED_DATE";
    }

    // 将POJO转换为对应15列的CSV行（从110个字段中提取需要的15个）
    public String convertPojoToCsvRow(YourEmployeePOJO pojo) {
        // 把POJO转成Map，方便提取字段
        Map<String, Object> fieldMap = objectMapper.convertValue(pojo, Map.class);
        
        // 按表头顺序提取字段值，注意和表头列数、顺序完全对应
        String[] values = {
                getFieldValue(fieldMap, "empId"),
                getFieldValue(fieldMap, "name"),
                getFieldValue(fieldMap, "department"),
                getFieldValue(fieldMap, "email"),
                getFieldValue(fieldMap, "phone"),
                getFieldValue(fieldMap, "hireDate"),
                getFieldValue(fieldMap, "salary"),
                getFieldValue(fieldMap, "managerId"),
                getFieldValue(fieldMap, "location"),
                getFieldValue(fieldMap, "jobTitle"),
                getFieldValue(fieldMap, "empType"),
                getFieldValue(fieldMap, "joinYear"),
                getFieldValue(fieldMap, "status"),
                getFieldValue(fieldMap, "country"),
                LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE) // 动态列值
        };
        
        // 用|拼接，处理空值
        return String.join("|", values);
    }

    // 生成尾记录（比如统计总条数+生成时间）
    public String generateFooter(int totalRecords) {
        return "TOTAL_RECORDS:" + totalRecords + "|FINISHED_AT:" + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
    }

    // 工具方法：安全获取字段值，空值返回空字符串
    private String getFieldValue(Map<String, Object> fieldMap, String fieldName) {
        Object value = fieldMap.get(fieldName);
        return value == null ? "" : value.toString();
    }
}

2. 在路由中整合聚合策略 #

因为你可能会处理多条POJO数据，所以用AggregationStrategy把所有行聚合起来，再加上头和尾：

import org.apache.camel.Exchange;
import org.apache.camel.processor.aggregate.AggregationStrategy;

from("http://localhost:8080/employeePayload")
        .unmarshal(yourExistingJsonDataFormat) // JSON转POJO
        .aggregate(constant(true), new AggregationStrategy() {
            private int recordCount = 0;
            private final StringBuilder csvContent = new StringBuilder();
            private final EmployeeCsvTransformer transformer = new EmployeeCsvTransformer();

            @Override
            public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                if (oldExchange == null) {
                    // 第一条数据：先写入头记录
                    csvContent.append(transformer.generateHeader()).append("\n");
                }

                // 转换当前POJO为CSV行
                YourEmployeePOJO pojo = newExchange.getIn().getBody(YourEmployeePOJO.class);
                csvContent.append(transformer.convertPojoToCsvRow(pojo)).append("\n");
                recordCount++;

                // 更新聚合后的内容
                newExchange.getIn().setBody(csvContent.toString());
                return newExchange;
            }

            @Override
            public void onCompletion(Exchange exchange) {
                // 最后写入尾记录
                String footer = transformer.generateFooter(recordCount);
                String finalCsv = exchange.getIn().getBody(String.class) + footer;
                exchange.getIn().setBody(finalCsv);
            }
        })
        .completionTimeout(5000) // 5秒超时（根据你的实际需求调整，比如按数量完成）
        .to("file:/your/output/path?fileName=employees_${date:now:yyyyMMddHHmmss}.csv");

1. 完全无需Getter/Setter：依赖Jackson的字段访问能力，不用修改POJO的任何代码。
2. 高度自定义：头/尾记录可以根据需求自由调整，静态/动态内容都能轻松实现。
3. 灵活扩展：如果后续需要调整列数或字段，只需要修改Transformer Bean里的逻辑即可。

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