继续 zip()，如果一个源完成

如何解决继续 zip()，如果一个源完成

我在使用 .zip() 运算符时遇到了一些麻烦。

让我用一个小例子来简化我的问题。

    Flux<Integer> flux1 = Flux.just(9,8,3,-2);
    Flux<Integer> flux2 = Flux.just(7);
    Flux<Integer> flux3 = Flux.just(6,5,4,-4);

    List<Flux<Integer>> list1 = Arrays.asList(flux1,flux2,flux3);

    TreeSet<Integer> set = new TreeSet<>(Comparator.reverSEOrder());

    Set<Integer> list = Flux.zip(list1,objects -> {
        boolean setChanged = false;
        for (Object o : objects) {
            Integer i = (Integer) o;

            if (set.size() < 5 || i > set.last()) {
                setChanged = true;
                set.add(i);
                if (set.size() > 5) {
                    set.pollLast();
                }
            }
        }

        return setChanged;
    }).takeWhile(val -> val)
            .then(Mono.just(set))
            .block();

    System.out.println(list);

这里我有 3 个不同的源（默认情况下它们按降序排序，而且它们的数量可能更大），我想从它们中获取 5 个按降序排序的元素的集合。不幸的是，我不能只使用 concat() 或 merge() 运算符，因为现实生活中的资源可能非常大，但我只需要少量元素。

我期待 [9,7,6,5] 在这里，但其中一个来源在第一次压缩迭代后完成。

你能建议我如何解决这个问题吗？

解决方法

你可以试试reduce操作

import React from "react";

import icon_branding from "../assets/images/icon_branding.svg";
import icon_web from "../assets/images/icon_web.svg";
import icon_data from "../assets/images/icon_data.svg";
import icon_software from "../assets/images/icon_software.svg"

const icons = [

  {id:1,name:'icon_branding',icon_img:icon_branding},{id:2,name:'icon_web',icon_img:icon_web},{id:3,name:'icon_data',icon_img:icon_data},{id:4,name:'icon_software',icon_img:icon_software}
];


const ServiceCard = ({ icon,title,text,link }) => {
 

  console.log('this is the icon',icon)
  return (
    <div className="single-service-one">
      <div className="hover-block"></div>
      <img className = "service_icon_img" 
      src={`../assets/images/${icon}.svg`}>

      </img>
      <h3>{title}</h3>
      <p>{text}</p>
      <div className="line-block"></div>
      <a href={link} className="more-link">
        Read More
      </a>
    </div>
  );
};

export default ServiceCard;

您可以分块获取数据并进行比较以找到前 K 个元素。

在顺序情况下，它将获取一个新批次，将其与当前前 k 个结果进行比较，并返回一个新的 topk，如上例所示（如果 k 很大，PriorityQueue 可能更适合排序）。 如果您使用并行调度程序并且批处理是并行获取的，那么它可以将它们相互独立地进行比较，这应该会更快一些。 您还可以通过 rateLimit、buffer、delayElements 等完全控制获取的数据