java - 添加新对象与添加已创建对象的优先级队列

Question

我正在编写一个mapreduce程序，其中reduce函数接收一个可迭代的PageRankNode(具有两个字段)对象作为输入值，并将其添加到优先级队列中。在迭代每个对象并将其添加到优先级队列时，生成的优先级队列仅包含我添加的最后一个对象。但是，当我创建相同类型的新对象并将其添加到优先级队列时，它似乎按预期工作。

我想知道为什么会发生这种情况？下面是示例作品。但是，我使用“topPages.add(pageNode)”而不是“topPages.add(new PageRankNode(pageNode.pageName,pageNode.pageRank))”，它无法按预期工作。

下面还添加了优先级队列的比较器实现。

private Comparator<PageRankNode> comparator= new PageNodeComparator();
    private PriorityQueue<PageRankNode> topPages= new PriorityQueue<PageRankNode>(100,comparator);

public void reduce(NullWritable key,Iterable<PageRankNode> pageNodes,Context context) throws IOException,InterruptedException{
    for(PageRankNode pageNode:pageNodes){
        //topPages.add(pageNode);
        topPages.add(new PageRankNode(pageNode.pageName,pageNode.pageRank));
        if(topPages.size()>100){
            topPages.poll();
        }
    }
    PageRankNode pageNode;
    while(!topPages.isEmpty()){
        pageNode=topPages.poll();
        context.write(NullWritable.get(),new Text(pageNode.pageName+":"+pageNode.pageRank));
    }

}
public class PageNodeComparator implements Comparator<PageRankNode>{

    public int compare(PageRankNode x,PageRankNode y){
        if(x.pageRank < y.pageRank){
            return -1;
        }
        if(x.pageRank > y.pageRank){
            return 1;
        }
        return 0;
    }
}

Answer 1

我认为您没有提供足够的信息来正确诊断此问题。我发现您的 reduce 方法中有 InterruptedException，这表明您可能在多个线程上运行它 - 如果是这样，这可能是根本原因。

我编写了一个小程序，执行相同的操作，其输出符合预期。

import java.util.Arrays;
import java.util.Comparator;
import java.util.PriorityQueue;

public class Main {
  private static Comparator<PageRankNode> comparator = new PageNodeComparator();
  private static PriorityQueue<PageRankNode> topPages = new PriorityQueue<PageRankNode>(100, comparator);

  public static void main(String[] args) {
    reduce(Arrays.asList(
      new PageRankNode("A", 1000),
      new PageRankNode("B", 1500),
      new PageRankNode("C", 500),
      new PageRankNode("D", 700),
      new PageRankNode("E", 7000),
      new PageRankNode("F", 60)
    ));
  }

  public static void reduce(Iterable<PageRankNode> pageNodes) {
    for(PageRankNode pageNode : pageNodes) {
      //topPages.add(pageNode);
      topPages.add(new PageRankNode(pageNode.pageName, pageNode.pageRank));
      if(topPages.size() > 100) {
        topPages.poll();
      }
    }
    PageRankNode pageNode;
    while(!topPages.isEmpty()) {
      pageNode = topPages.poll();
      System.out.println(pageNode.pageName);
    }
  }

  public static class PageRankNode {
    private String pageName;
    private int pageRank;

    public PageRankNode(String pageName, int pageRank) {
      this.pageName = pageName;
      this.pageRank = pageRank;
    }
  }

  public static class PageNodeComparator implements Comparator<PageRankNode> {

    @Override
    public int compare(PageRankNode x, PageRankNode y) {
      if(x.pageRank < y.pageRank) {
        return -1;
      }
      if(x.pageRank > y.pageRank) {
        return 1;
      }
      return 0;
    }
  }
}

输出是:

F
C
D
A
B
E