7.3　第二个例子：数据筛选算法

7.3　第二个例子：数据筛选算法

假设有大量描述某个项列表的数据。例如，假设有关于很多人的很多属性（姓名、姓氏、地址、电话号码等）。通常需要获得满足特定标准的数据。例如，想要获得在某一街道居住的人或者叫某个特定名字的人。

本节，你将实现这样一个筛选程序。我们采用了来自UCI的Census-Income KDD数据集，该数据集包含了1994年到1995年从美国人口普查局的人口普查结果中抽取的加权人口普查数据。

在本例的并发版本中，你将学会如何撤销在Fork/Join池中运行的任务，以及如何管理在任务中抛出的未校验异常。

7.3.1　公共特性

我们实现了一些类来读取文件数据并且进行筛选。这些类在算法的串行版本和并发版本中都会用到，具体如下。

CensusData类：该类存储了39个用于定义人员的属性。该类定义了这些属性以及获取和设置这些属性值的方法。我们将通过编号标识每个属性。该类的evaluateFilter()方法包含了属性名称与属性编号之间的关系。
CensusDataLoader类：该类从一个文件中加载人口普查数据。该类有一个load()方法，该方法将文件的路径作为输入参数，返回一个含有文件中所有人员信息的CensusData数组。
FilterData类：该类定义了一个数据筛选器。筛选器包括一个属性的编号和该属性的值。
Filter类：该类实现了一些方法来判定一个CensusData对象是否满足一个筛选器列表所设定的条件。

这里不再介绍这些类的源代码，它们都非常简单，你可以查看本例源代码的详情。

7.3.2　串行版

我们在两个类中实现了筛选算法的串行版本。SerialSearch类进行数据的筛选，该类提供了两个方法。

findAny()方法：该方法接收CensusData对象数组作为参数，其中有来自文件的数据和一个筛选器列表，而且该方法返回一个CensusData对象，其中含有第一个满足筛选器规定标准的人员。
findAll()方法：该方法接收CensusData对象数组作为参数，其中有来自文件的数据和一个筛选器列表，而且该方法返回一个CensusData对象数组，其中含有所有满足筛选器规定标准的人员。

SerialMain类实现了该版本程序的main()方法，并且进行程序测试，测量了该算法一些情况下的执行时间。

SerialSearch类

如前所述，该类实现了数据筛选功能。它提供了两个方法，第一个是findAny()方法，用于查找满足筛选器条件的第一个数据对象。该方法找到第一个数据对象时，其执行完成。相关代码如下：

public class SerialSearch {
  public static CensusData findAny (CensusData[] data, List<FilterData>
                                    filters) {
    int index=0;
    for (CensusData censusData : data) {
      if (Filter.filter(censusData, filters)) {
        System.out.println("Found: "+index);
        return censusData;
      }
      index++;
    }
    return null;
  }

第二个是findAll()方法，该方法返回一个CensusData对象数组，其中含有满足筛选标准的所有对象，具体如下：

  public static List<CensusData> findAll (CensusData[] data,
                                        List<FilterData> filters) {
    List<CensusData> results=new ArrayList<CensusData>();
    for (CensusData censusData : data) {
      if (Filter.filter(censusData, filters)) {
        results.add(censusData);
      }
    }
    return results;
  }
}

SerialMain类

你将在不同情况下使用该类测试筛选算法。首先，从文件加载数据，如下所示：

public class SerialMain {
  public static void main(String[] args) {
    Path path = Paths.get("data","census-income.data");
    CensusData data[]=CensusDataLoader.load(path);
    System.out.println("Number of items: "+data.length);
    Date start, end;

我们要做的第一件事是使用findAny()方法查找出现在数组中第一个位置的对象。可以构建一个筛选器列表，然后调用findAny()方法，该方法的参数为文件中的数据和筛选器列表。

List<FilterData> filters=new ArrayList<>();
FilterData filter=new FilterData();
filter.setIdField(32);
filter.setValue("Dominican-Republic");
filters.add(filter);
filter=new FilterData();
filter.setIdField(31);
filter.setValue("Dominican-Republic");
filters.add(filter);
filter=new FilterData();
filter.setIdField(1);
filter.setValue("Not in universe");
filters.add(filter);
filter=new FilterData();
filter.setIdField(14);
filter.setValue("Not in universe");
filters.add(filter);
start=new Date();
CensusData result=SerialSearch.findAny(data, filters);
System.out.println("Test 1 - Result: "+result
                   .getReasonForUnemployment());
end=new Date();
System.out.println("Test 1- Execution Time: "+(end.getTime()-
                    start.getTime()));

筛选器根据下述属性进行查找。

32：这是生父的国籍属性。
31：这是生母的国籍属性。
1：这是工作类别属性，其中一个可能值是Not in universe。
14：这是失业原因属性，其中一个可能值是Not in universe。
我们将按照如下方式测试其他用例。
使用findAny()方法查找出现在数组中最后一个位置的对象。
使用findAny()方法尝试查找某个并不存在的对象。
在错误情境中使用findAny()方法。
使用findAll()方法获取满足筛选器列表条件的所有对象。
在错误情境中使用findAll()方法。

7.3.3　并发版本

我们将在并发版本中引入更多要素。

任务管理器：使用Fork/Join框架时，从一个任务开始，并且将该任务分割成两个或者更多子任务，之后再一次次分割，直到问题达到你想要的规模为止。有些情况下，需要结束所有任务。例如，实现findAny()方法并且找到了一个满足所有条件的对象时，就不需要继续执行剩下的任务了。
用于实现findAny()方法的RecursiveTask类：该类是扩展了RecursiveTask类的IndividualTask类。
用于实现findAll()方法的RecursiveTask类：该类是扩展了RecursiveTask类的ListTask类。

下面详细了解一下这些类。

TaskManager类

我们将使用该类来控制任务的撤销。我们将在下述两种情况中撤销任务的执行。

正在执行findAny()操作并且找到了满足要求的对象。
正在执行findAny()或findAll()操作并且在某个任务中出现了一个未校验异常。
该类声明了两个属性：一个是用于存放所有待撤销任务的ConcurrentLinkedDeque，另一个是用于保证只有一个任务执行cancelTasks()方法的AtomicBoolean变量。使用AtomicBoolean变量确保所有任务都能以线程安全的方式访问它们的值。

public class TaskManager {
  private Set<RecursiveTask> tasks;
  private AtomicBoolean cancelled;
  public TaskManager() {
    tasks = ConcurrentHashMap.newKeySet();
    cancelled = new AtomicBoolean(false);
  }

该类定义了向ConcurrentLinkedDeque添加某个任务的方法，从ConcurrentLinkedDeque中删除某个任务的方法，以及撤销存放在ConcurrentLinkedDeque中所有任务的方法。要撤销这些任务，我们使用在ForkJoinTask类中定义的cancel()方法。该方法的参数为true时会强制中断运行中的任务，如下所示：

public void addTask(RecursiveTask task) {
  tasks.add(task);
}
public void cancelTasks(RecursiveTask sourceTask) {
  if (cancelled.compareAndSet(false, true)) {
    for (RecursiveTask task : tasks) {
      if (task != sourceTask) {
        if(cancelled.get()) {
          task.cancel(true);
        }
        else {
          tasks.add(task);
        }
      }
    }
  }
}
public void deleteTask(RecursiveTask task) {
  tasks.remove(task);
}

cancelTasks()方法接收一个RecursiveTask对象作为参数。除了调用该方法的任务之外，我们将撤销所有其他任务。我们不想撤销已经找到结果的任务。compareAndSet(false, true)方法将AtomicBoolean变量设置为true，而且当且仅当该变量的当前值为false时才会返回true值。如果AtomicBoolean变量已经有一个true值，那么该方法将返回false值。整个操作都以原子方式执行，因此即使cancelTasks()方法被不同线程同时调用多次，也能够保证if语句的主体部分最多执行一次。

IndividualTask类

IndividualTask类扩展了RecursiveTask类，以CensusData任务为参数，并且实现了findAny()操作。该类定义了如下属性。

一个含有所有CensusData对象的数组。
Start属性和end属性，它们确定了要处理的元素。
size属性，它确定了在无须分割任务的前提下所处理的最大元素数。
TaskManager类，它用于在必要之时撤销任务。
下面的代码给出了一个将要应用的筛选器列表。

private CensusData[] data;
private int start, end, size;
private TaskManager manager;
private List<FilterData> filters;
public IndividualTask(CensusData[] data, int start,
                      int end, TaskManager manager,
                      int size, List<FilterData> filters) {
  this.data = data;
  this.start = start;
  this.end = end;
  this.manager = manager;
  this.size = size;
  this.filters = filters;
}

该类中的主方法是compute()方法。该方法返回一个CensusData对象。如果任务需要处理的元素数比size属性值小，该方法直接进行对象查找。如果该方法找到了想要的对象，那么它将返回该对象并且使用cancelTasks()方法撤销剩余任务的执行。如果该方法没有找到想要的对象，那么它将返回null值，如下所示：

if (end - start <= size) {
  for (int i = start; i < end && ! Thread.currentThread()
       .isInterrupted(); i++) {
    CensusData censusData = data[i];
    if (Filter.filter(censusData, filters)) {
      System.out.println("Found: " + i);
      manager.cancelTasks(this);
      return censusData;
    }
  }
  return null;
}

如果要处理的项数要比size属性规定的多，那么要创建两个子任务来分别处理其中的一半元素。

} else {
  int mid = (start + end) / 2;
  IndividualTask task1 = new IndividualTask(data, start, mid, manager,
                                            size, filters);
  IndividualTask task2 = new IndividualTask(data, mid, end, manager,
                                            size, filters);

然后，向任务管理器添加新创建的任务，并且删除实际任务。如果要撤销任务，即指仅撤销正在运行的任务。

manager.addTask(task1);
manager.addTask(task2);
manager.deleteTask(this);

接着，使用fork()方法以异步方式将任务发送给ForkJoinPool，并且使用quietlyJoin()方法等待其执行结束。

join()方法和quietlyJoin()方法之间的区别在于，join()启动之后，如果任务撤销，将抛出异常，或者在方法内部抛出一个未校验异常，而quietlyJoin()方法则不抛出任何异常。

task1.fork();
task2.fork();
task1.quietlyJoin();
task2.quietlyJoin();

然后，从TaskManager类中删除子任务，如下所示：

manager.deleteTask(task1);
manager.deleteTask(task2);

现在，使用join()方法获取任务的结果。如果一个任务抛出一个未校验异常，那么该异常将被传播而无须特殊处理，而撤销操作则直接被忽略，如下所示：

    try {
      CensusData res = task1.join();
      if (res != null)
        return res;
        manager.deleteTask(task1);
      } catch (CancellationException ex) {
    }
    try {
      CensusData res = task2.join();
      if (res != null)
        return res;
        manager.deleteTask(task2);
      } catch (CancellationException ex) {
    }
    return null;
  }
}

ListTask类

ListTask类扩展了RecursiveTask类，采用一个CensusData对象列表作为参数。我们将使用该任务来实现findAll()操作。该任务和IndividualTask任务非常相似，都使用相同的属性，但是它们在compute()方法上有所不同。

首先，初始化一个List对象以返回结果并且校验任务要处理的元素数量。如果任务要处理的元素数量小于size属性，将满足筛选器指定标准的所有对象添加到结果列表中。

@Override
protected List<CensusData> compute() {
  List<CensusData> ret = new ArrayList<CensusData>();
  List<CensusData> tmp;
  if (end - start <= size) {
    for (int i = start; i < end; i++) {
      CensusData censusData = data[i];
      if (Filter.filter(censusData, filters)) {
        ret.add(censusData);
      }
    }

如果要处理的项数多于size属性，将创建两个子任务来处理其中各一半的元素。

int mid = (start + end) / 2;
ListTask task1 = new ListTask(data, start, mid, manager, size,
                              filters);
ListTask task2 = new ListTask(data, mid, end, manager, size, filters);

然后，将新创建的任务添加到任务管理器并且删除原来的实际任务。该实际任务并不会被撤销，而其子任务会被撤销，如下所示：

manager.addTask(task1);
manager.addTask(task2);
manager.deleteTask(this);

然后，使用fork()方法以异步方式将任务发送给ForkJoinPool，并且使用quietlyJoin()方法等待其执行结束。

task1.fork();
task2.fork();
task2.quietlyJoin();
task1.quietlyJoin();

然后，从TaskManager中删除子任务。

manager.deleteTask(task1);
manager.deleteTask(task2);

现在，使用join()方法获取任务结果。如果一个任务抛出了未校验异常，那么它将被传播而不经特殊处理，并且会直接忽略撤销操作。

    try {
      tmp = task1.join();
      if (tmp != null)
        ret.addAll(tmp);
        manager.deleteTask(task1);
      } catch (CancellationException ex) {
    }
    try {
      tmp = task2.join();
      if (tmp != null)
        ret.addAll(tmp);
        manager.deleteTask(task2);
      } catch (CancellationException ex) {
    }
  }
}

ConcurrentSearch类

ConcurrentSearch类实现了findAny()和findAll()方法。这两个方法与串行版本的相应方法有着相同的接口。从内部来看，它们初始化TaskManager对象和第一个任务，并且使用execute方法将其发送给默认的ForkJoinPool，然后等待任务结束并且输出结果。下面是findAny()方法的代码。

public class ConcurrentSearch {
  public static CensusData findAny (CensusData[] data,
                      List<FilterData> filters, int size) {
  TaskManager manager=new TaskManager();
  IndividualTask task=new IndividualTask(data, 0, data.length,
                                         manager, size, filters);
  ForkJoinPool.commonPool().execute(task);
  try {
    CensusData result=task.join();
    if (result!=null) {
      System.out.println("Find Any Result: "+result.getCitizenship());
      return result;
    } catch (Exception e) {
    System.err.println("findAny has finished with an error: "+
                       task.getException().getMessage());
  }
  return null;
}

下面是findAll()方法的代码。

public static CensusData[] findAll (CensusData[] data,
    List<FilterData> filters, int size) {
    List<CensusData> results;
    TaskManager manager=new TaskManager();
    ListTask task=new ListTask(data,0,data.length,manager,
                                size,filters);
    ForkJoinPool.commonPool().execute(task);
    try {
      results=task.join();
      return results;
    } catch (Exception e) {
    System.err.println("findAny has finished with an
                       error: " + task.getException().getMessage());
  }
  return null;
}

ConcurrentMain类

ConcurrentMain类用于测试目标筛选器的并发版本。它和SerialMain类似，只是采用了各种并发版本的操作。

7.3.4　对比两个版本

为了比较筛选算法的串行版本和并发版本，我们分六种情形进行测试。

测试1：测试findAny()方法，查找一个对象，它在CensusData数组中的第一个位置。
测试2：测试findAny()方法，查找一个对象，它在CensusData数组的最后一个位置。
测试3：测试findAny()方法，查找一个不存在的对象。
测试4：在错误情况下测试findAny()方法。
测试5：在正常情况下测试findAll()方法。
测试6：在错误情况下测试findAll()方法。

对于算法的并发版本，我们测试了size参数的3组不同取值，该参数确定了一个任务在不分割成两个子任务时所能处理的最大元素数。测试使用的最大阈值为10、200、2000和4000个元素。

采用JMH框架执行这些示例，该框架允许在Java中实现微型基准测试。使用面向基准测试的框架是比较好的解决方案，它直接用currentTimeMillis()方法或者nanoTime()方法度量时间。在两种不同的架构上分别执行这些示例10次。

一台计算机配置了Intel Core i5-5300处理器、Windows 7操作系统和16GB的RAM。该处理器有两个核，且每个核可以执行两个线程，这样就有四个并行线程。
另一台计算机配置了AMD A8-640处理器、Windows 10操作系统和8GB的RAM。该处理器有四个核。

与其他例子相同，以毫秒为单位度量执行时间。首先给出在AMD架构上的运行结果。

AMD架构
测试用例	串行版本	并发版本规模=10	并发版本规模=200	并发版本规模=2000	并发版本规模=4000	最优
测试1	2.374	8.041	5.434	4.802	9.339	串行版本
测试2	86.049	75.872	57.954	32.56	32.876	并发版本
测试3	58.322	70.562	22.947	30.831	27.033	并发版本
测试4	0.65	15 090.17	259.597	8.585	5.987	串行版本
测试5	60.129	42.979	44.81	22.741	21.287	并发版本
测试6	0.697	14 279.35	256.271	9.365	4.842	串行版本

在Intel架构上运行的结果如下。

Intel架构
测试用例	串行版本	并发版本规模=10	并发版本规模=200	并发版本规模=2000	并发版本规模=4000	最优
测试1	0.796	8.896	3.253	2.08	2.422	串行版本
测试2	31 006	41.312	32.974	14.407	14.55	并发版本
测试3	15.076	25.068	9.55	10.729	9.77	并发版本
测试4	0.378	10 664.607	106.349	4.699	2.898	串行版本
测试5	13.291	18.037	25.061	10.262	8.937	并发版本
测试6	0.352	10 901.387	91.998	5.246	2.24	串行版本

根据上述表格，可以得出如下结论。

处理相对少量的元素时，算法的串行版本具有更好的性能。
处理所有元素或者其中的一部分时，算法的并发版本具有更好的性能。
在错误情况下，算法的串行版本要比并发版本的性能更好。当size参数的值较小时，并发版本在这种情况下性能非常糟糕。

在这种情况下，并发处理并不会总能提升性能。

7.3 第二个例子：数据筛选算法