27、实战案例五：数据处理管道重构 - 提升可读性与可维护性

数据处理管道，说白了就是把一堆数据从一个形态变成另一个形态的流水线。我在项目中见过太多这样的代码——一个函数里塞了七八个步骤，每个步骤之间用临时变量传递，最后返回一个结果。嗯，这种代码能跑，但改起来真要命。

今天这个案例，是我从一个数据报表系统中抽出来的。当时接手的时候，整个管道有200多行，全写在一个方法里。我花了整整一个下午才理清楚它到底在干什么。重构之后，代码量没少多少，但可读性提升了不止一个档次。

重构前的代码长什么样

先看看原始代码。这是一个处理用户行为数据的管道，从原始日志到最终报表，中间经历了过滤、清洗、聚合、格式化四个阶段。

public class DataProcessor {
    public List<ReportRow> process(List<RawLog> logs) {
        // 第一步：过滤无效数据
        List<RawLog> validLogs = new ArrayList<>();
        for (RawLog log : logs) {
            if (log.getUserId() != null && log.getAction() != null
                    && log.getTimestamp() > 0) {
                validLogs.add(log);
            }
        }

        // 第二步：清洗数据，补全缺失字段
        List<CleanedLog> cleanedLogs = new ArrayList<>();
        for (RawLog log : validLogs) {
            CleanedLog cleaned = new CleanedLog();
            cleaned.setUserId(log.getUserId());
            cleaned.setAction(log.getAction());
            cleaned.setTimestamp(log.getTimestamp());
            if (log.getDuration() == null) {
                cleaned.setDuration(0);
            } else {
                cleaned.setDuration(log.getDuration());
            }
            cleanedLogs.add(cleaned);
        }

        // 第三步：按用户分组聚合
        Map<String, List<CleanedLog>> grouped = new HashMap<>();
        for (CleanedLog log : cleanedLogs) {
            grouped.computeIfAbsent(log.getUserId(), k -> new ArrayList<>()).add(log);
        }

        // 第四步：计算每个用户的统计指标
        List<ReportRow> result = new ArrayList<>();
        for (Map.Entry<String, List<CleanedLog>> entry : grouped.entrySet()) {
            ReportRow row = new ReportRow();
            row.setUserId(entry.getKey());
            row.setTotalActions(entry.getValue().size());
            row.setAvgDuration(entry.getValue().stream()
                    .mapToInt(CleanedLog::getDuration).average().orElse(0));
            result.add(row);
        }

        return result;
    }
}

这段代码的问题很明显：所有逻辑混在一起，没有边界。你想想看，如果哪天要改过滤规则，你得在200行代码里找到对应的for循环。如果哪天要加一个排序步骤，你都不知道该插在哪里。

重构思路：把管道拆成独立步骤

我个人习惯的做法是：先画图，再写代码。把数据处理流程可视化出来，你就能清楚地看到每个步骤的输入和输出是什么。

你看，这张图把整个流程拆成了四个清晰的步骤。每个步骤只做一件事，输入输出都很明确。这就是重构的目标——让代码结构跟流程图一样清晰。

重构后的代码

我建议的做法是：每个步骤封装成一个独立的方法，方法名直接说明它在做什么。这样读代码的人一眼就能看到整个管道的骨架。

public class DataProcessor {
    public List<ReportRow> process(List<RawLog> logs) {
        List<RawLog> validLogs = filterInvalidLogs(logs);
        List<CleanedLog> cleanedLogs = cleanLogs(validLogs);
        Map<String, List<CleanedLog>> grouped = groupByUser(cleanedLogs);
        return formatReport(grouped);
    }

    private List<RawLog> filterInvalidLogs(List<RawLog> logs) {
        return logs.stream()
                .filter(log -> log.getUserId() != null)
                .filter(log -> log.getAction() != null)
                .filter(log -> log.getTimestamp() > 0)
                .collect(Collectors.toList());
    }

    private List<CleanedLog> cleanLogs(List<RawLog> logs) {
        return logs.stream()
                .map(log -> {
                    CleanedLog cleaned = new CleanedLog();
                    cleaned.setUserId(log.getUserId());
                    cleaned.setAction(log.getAction());
                    cleaned.setTimestamp(log.getTimestamp());
                    cleaned.setDuration(log.getDuration() != null ? log.getDuration() : 0);
                    return cleaned;
                })
                .collect(Collectors.toList());
    }

    private Map<String, List<CleanedLog>> groupByUser(List<CleanedLog> logs) {
        return logs.stream()
                .collect(Collectors.groupingBy(CleanedLog::getUserId));
    }

    private List<ReportRow> formatReport(Map<String, List<CleanedLog>> grouped) {
        return grouped.entrySet().stream()
                .map(entry -> {
                    ReportRow row = new ReportRow();
                    row.setUserId(entry.getKey());
                    row.setTotalActions(entry.getValue().size());
                    row.setAvgDuration(entry.getValue().stream()
                            .mapToInt(CleanedLog::getDuration)
                            .average()
                            .orElse(0));
                    return row;
                })
                .collect(Collectors.toList());
    }
}

重构带来的好处

重构之后，代码的可读性和可维护性都有了质的提升。具体来说：

维度	重构前	重构后
可读性	200行代码混在一起，需要逐行阅读才能理解	主方法5行，每个步骤一目了然
可维护性	修改一个步骤可能影响其他步骤	每个步骤独立，修改互不影响
可测试性	只能测试整个process方法	每个步骤都可以单独写单元测试
可复用性	逻辑耦合在方法内部，无法复用	每个方法都可以被其他管道调用

进一步优化：引入管道模式

如果你的管道需要频繁调整步骤顺序，或者需要动态组合不同的步骤，可以考虑引入管道模式。不过我个人建议，先保持简单，等确实需要了再升级。

// 定义一个函数式接口
@FunctionalInterface
interface PipeStep<T, R> {
    R execute(T input);
}

// 管道类
public class DataPipeline {
    private List<PipeStep> steps = new ArrayList<>();

    public DataPipeline addStep(PipeStep step) {
        steps.add(step);
        return this;
    }

    public Object execute(Object input) {
        Object result = input;
        for (PipeStep step : steps) {
            result = step.execute(result);
        }
        return result;
    }
}

// 使用示例
DataPipeline pipeline = new DataPipeline();
pipeline.addStep((List<RawLog> logs) -> filterInvalidLogs(logs))
        .addStep((List<RawLog> logs) -> cleanLogs(logs))
        .addStep((List<CleanedLog> logs) -> groupByUser(logs))
        .addStep((Map<String, List<CleanedLog>> grouped) -> formatReport(grouped));

List<ReportRow> result = (List<ReportRow>) pipeline.execute(logs);

你看，管道模式让步骤的组合变得更加灵活。你可以随时插入、删除、替换某个步骤，而不需要修改任何现有代码。不过说实话，在大多数业务场景下，方法级别的拆分已经够用了。管道模式更适合那些需要动态配置管道的场景。

总结

数据处理管道的重构，核心就一句话：把一个大方法拆成多个小方法，每个方法只做一件事。这样做的好处是显而易见的——代码更容易读、更容易改、更容易测。

嗯，最后再啰嗦一句：重构不是炫技，而是为了让代码更清晰。如果你重构完之后，代码反而更难懂了，那说明你走偏了。保持简单，保持清晰，这才是重构的真谛。

---

公众号:蓝海资料掘金营，微信deep3321