如何用Java Stream合并从不同CSV文件读取的城市与国家数据？

问题背景 #

我需要从两个CSV文件中读取城市和国家数据，通过Java Stream合并结果，必须严格保持城市列表的原始顺序。原本考虑过使用并行流或CompletableFuture，但由于第二个查询需要依赖第一个查询的结果（城市的国家代码），不确定这类方案是否适用。

核心需求是通过国家代码关联两个数据集，只为每个城市匹配对应的国家名称。例如城市列表为[Berlin, Kopenhag, Paris]，合并后的对应国家结果需为[Germany, Denmark, France]。

数据集结构 #

城市CSV

id   | name     | countryCode |
------------------------------
1    | Berlin   | DE          |
2    | Munich   | DE          |
3    | Köln     | DE          |
4    | Paris    | FR          |
5    | Kopenhag | DK          |
...

国家CSV

id     | name      | code        |
---------------------------------- 
100    | Germany   | DE          |
105    | France    | FR          |
108    | Denmark   | DK          |
...

实体类定义 #

City类

public class City{

    @CsvBindByPosition(position = 0)
    private Integer id;

    @CsvBindByPosition(position = 1)
    private String name;

    @CsvBindByPosition(position = 2)
    private String countryCode;

    // 省略getter、setter、toString方法
}

Country类

public class Country {

    @CsvBindByPosition(position = 0)
    private Integer id;

    @CsvBindByPosition(position = 1)
    private String name;

    @CsvBindByPosition(position = 2)
    private String code;

    // 省略getter、setter、toString方法
}

解决方案 #

核心思路 #

由于需要严格保持城市顺序，且第二个查询依赖第一个的结果，最优流程为：

1. 先读取所有城市数据，天然保留CSV的原始行顺序
2. 从城市列表中提取所有唯一的国家代码，批量读取对应国家数据并构建国家代码→国家名称的映射表（避免重复读取国家数据，提升效率）
3. 遍历城市列表，通过映射表匹配对应国家名称，完全保持原顺序

为什么不使用并行流或CompletableFuture？

• 并行流会打乱原始顺序，即使使用forEachOrdered也会增加不必要的复杂度，且无性能优势
• CompletableFuture适用于异步独立任务，但此处第二个任务依赖第一个的结果，异步执行反而会增加等待逻辑，不如同步处理直接高效

代码实现 #

首先引入OpenCSV依赖（适配实体类中的注解），Maven依赖如下：

<dependency>
    <groupId>com.opencsv</groupId>
    <artifactId>opencsv</artifactId>
    <version>5.6</version>
</dependency>

完整处理代码：

import com.opencsv.bean.CsvToBeanBuilder;
import java.io.FileReader;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class CityCountryMerger {

    public static void main(String[] args) throws Exception {
        // 1. 读取城市列表，保持原始顺序
        List<City> cities = new CsvToBeanBuilder<City>(new FileReader("cities.csv"))
                .withType(City.class)
                .build()
                .parse();

        // 2. 提取所需国家代码，构建国家代码与名称的映射表
        List<String> requiredCountryCodes = cities.stream()
                .map(City::getCountryCode)
                .distinct()
                .collect(Collectors.toList());

        Map<String, String> countryCodeMap = new CsvToBeanBuilder<Country>(new FileReader("countries.csv"))
                .withType(Country.class)
                .build()
                .parse()
                .stream()
                .filter(country -> requiredCountryCodes.contains(country.getCode()))
                .collect(Collectors.toMap(Country::getCode, Country::getName));

        // 3. 合并数据，生成城市-国家对应列表，保持原顺序
        List<String> mergedResult = cities.stream()
                .map(city -> String.format("%s -> %s", city.getName(), countryCodeMap.get(city.getCountryCode())))
                .collect(Collectors.toList());

        // 输出结果
        mergedResult.forEach(System.out::println);
    }
}

扩展说明 #

如果需要返回自定义合并对象（如包含城市和国家信息的DTO），可修改最后一步的map操作：

// 假设已定义CityCountryDTO类
List<CityCountryDTO> mergedDTOList = cities.stream()
        .map(city -> {
            CityCountryDTO dto = new CityCountryDTO();
            dto.setCityName(city.getName());
            dto.setCountryName(countryCodeMap.get(city.getCountryCode()));
            return dto;
        })
        .collect(Collectors.toList());

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