由于没有人提供符合原始问题陈述所需的自定义收集器解决方案,因此这里提供了一个基于提供的时间间隔对Post
对象进行分组的收集器实现。
问题中提到的Date
类已经过时,不建议在新项目中使用。因此,将使用LocalDateTime
代替。
帖子和帖子组
为了测试目的,我使用了作为Java 16 记录实现的Post
(如果您将其替换为类,则整体解决方案将完全符合Java 8):
public record Post(LocalDateTime dateTime) {}
此外,我增强了
PostGroup
对象。我的想法是它应该能够决定是否将提供的
Post
添加到
帖子列表中,或者根据
信息专家原则拒绝它(简而言之:所有与数据的操作都应该发生在该数据所属的类内部)。
为了实现这个功能,添加了两个额外的字段:
interval
类型为
Duration
,来自
java.time
包,用于表示一个
组 中
最早的帖子 和
最新的帖子 之间的最大
时间间隔,以及类型为
LocalDateTime
的
intervalBound
字段。该字段在第一个
帖子 添加后进行初始化,并将在方法
isWithinInterval()
内部使用,以检查所提供的
帖子 是否适合于
时间间隔。
public class PostsGroup {
private Duration interval;
private LocalDateTime intervalBound;
private List<Post> posts = new ArrayList<>();
public PostsGroup(Duration interval) {
this.interval = interval;
}
public boolean tryAdd(Post post) {
if (posts.isEmpty()) {
intervalBound = post.dateTime().plus(interval);
return posts.add(post);
} else if (isWithinInterval(post)) {
return posts.add(post);
}
return false;
}
public boolean isWithinInterval(Post post) {
return post.dateTime().isBefore(intervalBound);
}
@Override
public String toString() {
return "PostsGroup{" + posts + '}';
}
}
我有两个假设:
- 来源中所有帖子都按时间排序(如果不是,请在收集结果之前介绍
sorted()
操作);
- 需要将帖子收集到最少的组中,因此不可能将此任务拆分并并行执行。
构建自定义收集器
我们可以通过使用静态方法Collector.of()
的某个版本或定义实现Collector
接口的类来创建一个自定义收集器。
创建自定义收集器时必须提供这些参数:
供应商 Supplier<A>
的目的是提供一个可变容器来存储流中的元素。在这种情况下,ArrayDeque
(作为Deque
接口的实现)将成为一个方便访问最近添加的元素(即最新的PostGroup
)的容器。
累加器 BiConsumer<A,T>
定义了如何将元素添加到供应商提供的容器中。对于此任务,我们需要提供逻辑,以允许确定来自流的下一个元素(即下一个Post
)是否应该进入在Deque
中的最后一个PostGroup
,或者需要为其分配新的PostGroup
。
组合器 BinaryOperator<A> combiner()
建立了一条规则,用于合并在并行执行流时获得的两个容器。由于该操作被视为不可并行化,组合器被实现为在并行执行时抛出AssertionError
。
完成函数 Function<A,R>
的目的是通过转换可变容器来生成最终结果。下面代码中的完成函数将包含结果的deque转换为一个不可变列表。
注意:Java 16方法toList()
在finisher函数中使用,对于Java 8,可以用collect(Collectors.toUnmodifiableList())
或collect(Collectors.toList())
替换。
- 特征允许提供额外的信息,例如
Collector.Characteristics.UNORDERED
,在这种情况下表示在并行执行时产生的减少部分结果的顺序不重要。在这种情况下,collector不需要任何特征。
下面的方法负责根据提供的interval生成collector。
public static Collector<Post, ?, List<PostsGroup>> groupPostsByInterval(Duration interval) {
return Collector.of(
ArrayDeque::new,
(Deque<PostsGroup> deque, Post post) -> {
if (deque.isEmpty() || !deque.getLast().tryAdd(post)) {
PostsGroup postsGroup = new PostsGroup(interval);
postsGroup.tryAdd(post);
deque.addLast(postsGroup);
}
},
(Deque<PostsGroup> left, Deque<PostsGroup> right) -> { throw new AssertionError("should not be used in parallel"); },
(Deque<PostsGroup> deque) -> deque.stream().collect(Collectors.collectingAndThen(Collectors.toUnmodifiableList())));
}
main()
- 演示
public static void main(String[] args) {
List<Post> posts =
List.of(new Post(LocalDateTime.of(2022,4,28,15,0)),
new Post(LocalDateTime.of(2022,4,28,15,3)),
new Post(LocalDateTime.of(2022,4,28,15,5)),
new Post(LocalDateTime.of(2022,4,28,15,8)),
new Post(LocalDateTime.of(2022,4,28,15,12)),
new Post(LocalDateTime.of(2022,4,28,15,15)),
new Post(LocalDateTime.of(2022,4,28,15,18)),
new Post(LocalDateTime.of(2022,4,28,15,27)),
new Post(LocalDateTime.of(2022,4,28,15,48)),
new Post(LocalDateTime.of(2022,4,28,15,54)));
Duration interval = Duration.ofMinutes(10);
List<PostsGroup> postsGroups = posts.stream()
.collect(groupPostsByInterval(interval));
postsGroups.forEach(System.out::println);
}
输出:
PostsGroup{[Post[dateTime=2022-04-28T15:00], Post[dateTime=2022-04-28T15:03], Post[dateTime=2022-04-28T15:05], Post[dateTime=2022-04-28T15:08]]}
PostsGroup{[Post[dateTime=2022-04-28T15:12], Post[dateTime=2022-04-28T15:15], Post[dateTime=2022-04-28T15:18]]}
PostsGroup{[Post[dateTime=2022-04-28T15:27]]}
PostsGroup{[Post[dateTime=2022-04-28T15:48], Post[dateTime=2022-04-28T15:54]]}
您也可以尝试一下这个在线演示