并行搜索

Posted 2021-01-13 joe-go

　　搜索是几乎每一个软件都必不可少的功能。对于有序数据，通常采用二分查找法。对于无序数据，则只能挨个查找。

　　给定一个数组，我们要查找满足条件的元素。对于串行程序来说，只要遍历一下数组就可以得到结果。但是如果要使用并行方式，则需要额外增加一些线程间的通信机制，使各个线程可以有效的运行。

　　一种简单的策略就是将原始数据集合按照期望的线程数进行分割。如果我们计划使用两个线程进行搜索，那么就可以把一个数组或者集合分割成两个。每个线程进行独立的搜索，当其中一个线程找到数据后，立即返回结果即可。

　　假设有一个整数数组，我们需要查找数组内的元素：

 1 public class CurrentSearch {
 2     static int[] arr = {3,2,32,323123,435,45,768,79,67,789,8098,0};//定义整型数组
 3     static ExecutorService pool = Executors.newCachedThreadPool();
 4     static final int Thread_Num = 2;
 5     static AtomicInteger result = new AtomicInteger(-1);//定义存放结果的变量，-1表示没有找到给定元素。
 6 
 7     /**
 8      * 并发搜索会要求每个线程查找arr中的一段，因此搜索方法必须指定线程需要搜索的开始位置和结束位置
 9      */
10     public static int search(int searchValue,int beginPos,int endPos){
11         int i = 0;
12         for (i = beginPos;i < endPos;i++){
13             if (result.get() >= 0){//判断是否已经有其他线程搜索到了需要的结果
14                 return result.get();
15             }
16             if (arr[i] == searchValue){
17                 //如果设置失败，表明其他线程已经先找到了
18                 if (!result.compareAndSet(-1,i)){//使用CAS操作，可以无视失败的情况
19                     return result.get();
20                 }
21                 return i;
22             }
23         }
24         return -1;
25     }
26     //定义一个线程查找，调用前面的search()方法
27     public static class SearchTask implements Callable<Integer>{
28 
29         int begin,end,searchValue;
30 
31         public SearchTask(int searchValue,int begin,int end){
32             this.begin = begin;
33             this.end = end;
34             this.searchValue = searchValue;
35         }
36         @Override
37         public Integer call() throws Exception {
38             int re = search(searchValue,begin,end);
39             return re;
40         }
41     }
42     //根据线程数量进行划分，并建立对应的任务提交给线程池处理
43     public static int pSearch(int searchValue) throws ExecutionException, InterruptedException {
44         int subArrSize = arr.length/Thread_Num+1;
45         List<Future<Integer>> re = new ArrayList<Future<Integer>>();
46         for (int i = 0;i < arr.length;i += subArrSize){
47             int end = i + subArrSize;
48             if (end >= arr.length){
49                 end = arr.length;
50             }
51             re.add(pool.submit(new SearchTask(searchValue,i,end)));
52         }
53         for (Future<Integer> future: re){
54             if (future.get() >= 0){
55                 return future.get();
56             }
57         }
58         return -1;
59     }
60     //测试
61     public static void main(String[] args) throws ExecutionException, InterruptedException {
62         int value = pSearch(0);
63         System.out.println(value);
64     }
65 }

 上述代码使用了Future模式，其中第46到52行，将原始数组划分成若干份，并根据每段结果建立子任务（每段分配一个线程），每一个子任务都会返回一个Future对象，通过Future对象可以获得线程组得到的最终结果，在这个例子中，我们通过result共享彼此的信息，因此只要当一个线程成功返回后，其他线程就会立即返回（代码第18到20行）。因此，不会出现由于排在前面的任务长时间无法结束而导致整个搜索结果无法立即获取的情况。

参考：《Java高并发程序设计》 葛一鸣 郭超 编著：