在处理金融或交易数据时，一个常见的需求是，针对特定实体（如货币）在每天的记录中，只保留最新（即时间戳最大）的一条。例如，我们可能有一系列关于某种货币在不同时间点接收到的数据，但我们只关心每天的最终更新状态。
假设我们有以下 Currency 类定义：
import java.time.LocalDateTime;
import java.time.LocalDate; // 用于日期部分
class Currency {
private Integer id;
private String name;
private LocalDateTime lastReceived;
public Currency(Integer id, String name, LocalDateTime lastReceived) {
this.id = id;
this.name = name;
this.lastReceived = lastReceived;
}
public Integer getId() { return id; }
public String getName() { return name; }
public LocalDateTime getLastReceived() { return lastReceived; }
@Override
public String toString() {
return "Currency{" +
"id=" + id +
", name='" + name + '\'' +
", lastReceived=" + lastReceived +
'}';
}
}以及以下示例数据：
| ID | NAME | LAST_RECEIVED |
|---|---|---|
| 1 | USD | 2022-05-18 09:04:01.545 |
| 2 | USD | 2022-05-18 08:04:01.545 |
| 3 | USD | 2022-05-19 08:04:01.545 |
| 4 | USD | 2022-05-20 08:04:01.545 |
| 5 | USD | 2022-05-20 11:04:01.545 |
| 6 | BUSD | 2022-05-18 08:04:01.545 |
我们的目标是获取如下结果：
| ID | NAME | LAST_RECEIVED |
|---|---|---|
| 1 | USD | 2022-05-18 09:04:01.545 |
| 3 | USD | 2022-05-19 08:04:01.545 |
| 5 | USD | 2022-05-20 11:04:01.545 |
可以看到，对于每天的USD记录，我们只保留了时间戳最晚的那一条。
立即学习“Java免费学习笔记（深入）”；
一、使用Java 8 Stream API实现
Java 8的Stream API提供了强大的数据处理能力，我们可以利用 groupingBy 和 collectingAndThen 组合来解决这个问题。核心思路是：首先根据货币名称和日期对数据进行分组，然后在每个组内找到时间戳最大的记录。
1.1 获取所有货币在每天的最新记录
为了实现对所有货币的每日最新记录提取，我们需要一个复合键来分组：货币名称和 lastReceived 的日期部分。
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
public class CurrencyProcessor {
public static void main(String[] args) {
List<Currency> data = Arrays.asList(
new Currency(1, "USD", LocalDateTime.parse("2022-05-18T09:04:01.545")),
new Currency(2, "USD", LocalDateTime.parse("2022-05-18T08:04:01.545")),
new Currency(3, "USD", LocalDateTime.parse("2022-05-19T08:04:01.545")),
new Currency(4, "USD", LocalDateTime.parse("2022-05-20T08:04:01.545")),
new Currency(5, "USD", LocalDateTime.parse("2022-05-20T11:04:01.545")),
new Currency(6, "BUSD", LocalDateTime.parse("2022-05-18T08:04:01.545")),
new Currency(7, "EUR", LocalDateTime.parse("2022-05-18T10:00:00.000")),
new Currency(8, "EUR", LocalDateTime.parse("2022-05-18T09:00:00.000"))
);
// 1. 获取所有货币在每天的最新记录 (无序)
List<Currency> lastByDate = new ArrayList<>(data
.stream() // Stream<Currency>
.collect(Collectors.groupingBy(
// 分组键：货币名称 + 日期 (LocalDate)
curr -> Arrays.asList(curr.getName(), curr.getLastReceived().toLocalDate()),
// 在每个组内，找到lastReceived最大的Currency对象
Collectors.collectingAndThen(
Collectors.maxBy(Comparator.comparing(Currency::getLastReceived)),
Optional::get // 将Optional<Currency>转换为Currency
)
)) // 结果是一个 Map<List<Object>, Currency>
.values() // 获取Map中所有的Currency值，形成一个Collection
);
System.out.println("所有货币在每天的最新记录 (无序):");
lastByDate.forEach(System.out::println);
// 2. 如果需要排序，可以对结果列表进行排序
lastByDate.sort(
Comparator.comparing(Currency::getName)
.thenComparing(Currency::getLastReceived)
);
System.out.println("\n所有货币在每天的最新记录 (按名称和时间排序):");
lastByDate.forEach(System.out::println);
}
}代码解析：
- Collectors.groupingBy(keyMapper, valueCollector): 这是核心操作。
- keyMapper: curr -> Arrays.asList(curr.getName(), curr.getLastReceived().toLocalDate()) 创建了一个复合键，确保我们按货币名称和日期进行分组。toLocalDate() 方法将 LocalDateTime 转换为 LocalDate，只保留日期部分。
- valueCollector: Collectors.collectingAndThen(Collectors.maxBy(Comparator.comparing(Currency::getLastReceived)), Optional::get)。
- Collectors.maxBy(Comparator.comparing(Currency::getLastReceived)): 在每个分组内部，找到 lastReceived 值最大的 Currency 对象。这个操作返回一个 Optional
。 - Optional::get: 由于我们知道每个分组至少包含一个元素（否则它不会被创建），所以 maxBy 总是会找到一个值，因此可以直接使用 Optional::get 来提取 Currency 对象。在实际应用中，如果分组可能为空，应使用 orElse 或 orElseThrow 进行更安全的处理。
- Collectors.maxBy(Comparator.comparing(Currency::getLastReceived)): 在每个分组内部，找到 lastReceived 值最大的 Currency 对象。这个操作返回一个 Optional
- .values(): 从 Map 中获取所有值（即我们筛选出的 Currency 对象），形成一个 Collection。
- new ArrayList(...): 将 Collection 转换为 ArrayList，以便后续进行排序。
- lastByDate.sort(...): 如果需要对最终结果进行排序，可以根据 name 和 lastReceived 进行多字段排序。
1.2 获取特定货币在每天的最新记录
如果我们的需求是针对某个特定的货币（例如 "USD"），那么我们可以先进行过滤，然后简化分组键。
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
public class CurrencyProcessorSpecific {
public static void main(String[] args) {
List<Currency> data = Arrays.asList(
new Currency(1, "USD", LocalDateTime.parse("2022-05-18T09:04:01.545")),
new Currency(2, "USD", LocalDateTime.parse("2022-05-18T08:04:01.545")),
new Currency(3, "USD", LocalDateTime.parse("2022-05-19T08:04:01.545")),
new Currency(4, "USD", LocalDateTime.parse("2022-05-20T08:04:01.545")),
new Currency(5, "USD", LocalDateTime.parse("2022-05-20T11:04:01.545")),
new Currency(6, "BUSD", LocalDateTime.parse("2022-05-18T08:04:01.545"))
);
String targetCurrency = "USD";
List<Currency> lastUSDByDate = new ArrayList<>(data
.stream()
.filter(curr -> targetCurrency.equalsIgnoreCase(curr.getName())) // 过滤特定货币
.collect(Collectors.groupingBy(
curr -> curr.getLastReceived().toLocalDate(), // 分组键现在只需日期
Collectors.collectingAndThen(
Collectors.maxBy(Comparator.comparing(Currency::getLastReceived)),
Optional::get
)
))
.values()
);
// 对结果按时间排序
lastUSDByDate.sort(Comparator.comparing(Currency::getLastReceived));
System.out.println("\n特定货币 (" + targetCurrency + ") 在每天的最新记录:");
lastUSDByDate.forEach(System::out::println);
}
}代码解析：
- filter(curr -> targetCurrency.equalsIgnoreCase(curr.getName())): 在分组之前，先过滤出我们感兴趣的货币。
- curr -> curr.getLastReceived().toLocalDate(): 过滤后，分组键就只需要 lastReceived 的日期部分了。
- 其余逻辑与之前相同。
二、使用原生SQL查询（窗口函数）
对于存储在数据库中的数据，使用原生SQL查询通常是更高效和更推荐的方法，特别是当数据集非常大时。SQL中的窗口函数（Window Functions）非常适合解决这类“每个分组内取Top N”的问题。
以PostgreSQL为例，我们可以使用 ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...) 来实现。
SELECT id, name, last_received
FROM (
SELECT
c.*,
ROW_NUMBER() OVER (
PARTITION BY name, to_char(last_received, 'yyyy-MM-dd') -- 按货币名称和日期分区
ORDER BY last_received DESC -- 在每个分区内按时间倒序排序
) AS rn -- 为每行分配一个行号
FROM Currency c
WHERE c.name = :currName -- 如果需要筛选特定货币
) tbl
WHERE rn = 1 -- 只保留每个分区的第一行（即时间最新的记录）
ORDER BY last_received; -- 最后按时间排序SQL查询解析：
-
内层查询 (子查询 tbl):
- ROW_NUMBER() OVER (...): 这是一个窗口函数。
- PARTITION BY name, to_char(last_received, 'yyyy-MM-dd'): 这定义了窗口（分组）。它告诉数据库将数据按 name 和 last_received 的日期部分进行分组。to_char(last_received, 'yyyy-MM-dd') 用于从时间戳中提取日期字符串。
- ORDER BY last_received DESC: 在每个窗口（分组）内部，数据会根据 last_received 字段进行降序排序。这意味着最新的时间戳会排在最前面。
- AS rn: ROW_NUMBER() 会为每个窗口中的每一行分配一个从1开始的连续整数。因为我们是降序排序，所以时间戳最新的那一行会得到 rn = 1。
-
外层查询:
- WHERE rn = 1: 这一步是关键，它筛选出每个分组中 rn 为1的记录，即每个货币在每天的最新记录。
- ORDER BY last_received: 对最终结果进行排序，使其按时间顺序显示。
- WHERE c.name = :currName: 如果只查询特定货币，可以在内层查询中添加此条件，这样可以减少处理的数据量，提高效率。
JPA集成注意事项：
目前，JPA（Java Persistence API）标准通常不支持直接在JPQL中使用窗口函数。因此，如果你正在使用JPA，你可能需要通过以下方式来执行此类查询：
- 原生查询（Native Query）: 如上述示例所示，使用 @Query(nativeQuery = true, value = "...") 注解在你的Repository接口中定义一个方法。
- 存储过程或视图: 将复杂的逻辑封装在数据库的存储过程或视图中，JPA可以直接调用存储过程或查询视图。
- MyBatis/JOOQ等其他ORM框架: 这些框架通常提供更灵活的SQL构建能力，可以更好地支持高级SQL特性。
三、总结与注意事项
- Java 8 Stream API: 适用于内存中的小型到中型数据集。它提供了简洁、函数式的编程风格。需要注意的是，当数据量非常大时，在JVM内存中处理所有数据可能会导致性能瓶颈或内存溢出。
- SQL窗口函数: 适用于数据库中存储的大型数据集。数据库引擎针对这类操作进行了高度优化，通常效率更高。对于复杂的聚合和排名需求，SQL是更强大的选择。
- 性能考量: 对于生产环境中的大量数据，优先考虑在数据库层面解决问题，利用数据库的索引和优化器。Java Stream API更适合对已经加载到内存中的数据进行二次处理。
- Optional::get 的安全性: 在Java Stream API的示例中，我们使用了 Optional::get。这假设每个分组至少包含一个元素。如果你的数据源可能导致某个分组为空（例如，如果 maxBy 的输入流为空），那么 Optional::get 会抛出 NoSuchElementException。在更严谨的代码中，你应该使用 orElse(null) 或 orElseThrow(...) 来处理 Optional 为空的情况。
- 日期处理: 在Java中，LocalDateTime.toLocalDate() 是提取日期部分的标准方法。在SQL中，不同的数据库有不同的函数（例如PostgreSQL的 to_char(date_column, 'yyyy-MM-dd') 或 date_trunc('day', date_column)，MySQL的 DATE(date_column)）。
选择哪种方法取决于你的数据量、数据存储位置以及对性能和代码可维护性的要求。通常，数据库原生查询是处理大规模数据这类问题的首选方案。
