对无序数组的并发搜索的java实现

Posted 2020-09-17 起个po名真费劲

对无序数组的并发搜索的实现可以充分的用到多cpu的优势

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

首先index需要采用atomicinteger来进行修饰，默认初始化的值为-1，意义为当前未找到，由于内部采用CAS机制，线程在遍历比较是否相等之前，会通过atomicinteger中的get方法拿到当前的值，如果大于等于0，那么说明别的线程已经找到了结果，直接返回get值就可以。如果比较的过程中发现相等了，那么调用atomicinteger中的compareAndSet(-1，i)，如果方法返回成功，则说明当前的线程是第一个发现结果的，那么返回当前index即可，如果失败，则说明别的线程先获得了结果，直接返回atomicinteger中的get方法获取的值即可。

整个过程采用future实现，拿到了Future后，不断地轮询结果，如果大于0即返回结果。

具体实现：

package parallel;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;


public class SeatchTask implements Callable<Integer>{
    static int[] arr = {2,34,5,6};
    static ExecutorService pool = Executors.newCachedThreadPool();
    static final int Thread_Num = 2;
    static AtomicInteger result = new AtomicInteger(-1);
    int begin,end,searchValue;
    public static int search(int searchValue, int beginPos, int endPos){
        int i = 0;
        for(i = beginPos; i < endPos;i ++){
            if(result.get() > 0){
                return result.get();
            }
            if(arr[i] == searchValue){
                if(!result.compareAndSet(-1, i)){
                    return result.get();
                }
                return i;
            }
        }
        return -1;
    }
    @Override
    public Integer call() throws Exception {
        int re = search(searchValue, begin, end);
        return re;
    }
    public SeatchTask(int searchValue, int begin, int end){
        this.searchValue = searchValue;
        this.begin = begin;
        this.end = end;
    }
    public static int pSearch(int searchValue) throws InterruptedException, ExecutionException{
        int subArrSize = arr.length/Thread_Num + 1;
        List<Future<Integer>> re = new ArrayList<Future<Integer>>();
        for(int i = 0;i < arr.length;i +=subArrSize){
            int end = i + subArrSize;
            if(end <= arr.length) end = arr.length;
            re.add(pool.submit(new SeatchTask(searchValue, i, end)));
        }
        for(Future<Integer> fu : re){
            if(fu.get() >= 0){
                return fu.get();
            }
        }
        return -1;
    }
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        int index = pSearch(34);
        System.out.println(index);
    }
}