第九章 - 并行数据结构 - Harvey Sydney

harveysydney

浏览: 10483 次
性别:
来自: 悉尼

最近访客更多访客>>

123yxc

youcp999

jeneo

fzxzoe

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

第九章 - 并行数据结构

博客分类：

Java多线程

在并行系统中，为了避免数据竞赛的产生，我们可以：

使用非同步的数据结构并自行添加同步机制代码
使用Java concurrency API 提供的数据结构

通常第二种方法是我们所推荐的。因为Java对这些数据结构已经做了优化

阻塞和非阻塞数据结构

Java concurrency API 提供了两种并行数据结构：

阻塞数据结构
非阻塞数据结构

并行数据结构实现的接口

BlockingQueue

Queue是一个线性数据结构，允许你往队列尾部插入数据并从头部取出数据。它是一个先进先出的数据结构。

Queue接口提供了以下方法：

往队列尾部插入元素
从队列头部读取元素并删除元素
从队列头部读取元素但不删除元素

接口定义了这些方法的两种版本，这两个版本在操作无法满足时具有不同的行为：

方法抛出异常
方法返回特殊的值，如 false 或 null

以下表格列出具有这些不同行为的方法

Operation	Exception	Special Value
Insert	add()	offer()
Retrieve and remove	remove()	poll()
Retrieve but don't remove	element()	peek()

BlockingDeque 接口继承了 Queue 接口，并添加了当操作无法满足时阻塞调用线程的方法，这些方法有：

Operation	Blocks
Insert	put()
Retrieve and remove	take()
Retrieve without removing	N/A

BlockingDeque

Deque 和 queue 一样，但允许你往两头添加和删除元素。Deque 接口继承了 Queue 接口。除了 Queue 接口提供的方法外，它提供了往队列两头添加，读取并删除，读取但不删除等方法：

Operation	Exception	Special Value
Insert	addFirst(), addLast()	offerFirst(), offerLast()
Retrieve and remove	removeFirst(), removeLast()	pollFirst(), pollLast()
Retrieve without removing	getFirst(), getLast()	peekFirst(), peekLast()

以下方法当操作无法满足时将阻塞调用线程：

Operation	Blocks
Insert	putFirst(), putLast()
Retrieve and remove	takeFirst(), takeLast()
Retrieve without removing	N/A

ConcurrentMap

Map 接口在 Java 8 里提供了一些新的方法：

forEach()：该方法对每一个元素调用参数里的函数
compute(), computeIfAbsent(), 和 computeIfPresent()：这些方法允许你传递一个方法来计算和key关联的值
merge()：该方法允许你把一对键值合并到map里，当键名不在map里时，直接插入键值。如果存在则调用参数里的方法来生成新的值

ConcurrentMap 继承了 Map 接口提供了并行操作

TransferQueue

该接口继承了 BlockingQueue 接口并添加了从生产者线程传递数据给消费者线程的方法。生产者线程可以等待直到消费者线程取走了队列中的元素。接口定义了如下方法：

transfer()：生产者把数据放入队列中然后生产者线程阻塞直到消费者线程把队列中的元素取走
tryTransfer()：如果有消费者线程正在等待读取数据，生产者则传递数据给消费者，否则该方法返回 fase 值，且数据不会被添加到队列中。

// 生产者线程
class Producer implements Runnable {
    private TransferQueue<String> transferQueue;
  
    private String name;
  
    private Integer numberOfMessagesToProduce;
  
    public AtomicInteger numberOfProducedMessages
      = new AtomicInteger();
 
    @Override
    public void run() {
        for (int i = 0; i < numberOfMessagesToProduce; i++) {
            try {
                boolean added 
                  = transferQueue.tryTransfer("A" + i, 4000, TimeUnit.MILLISECONDS);
                if(added){
                    numberOfProducedMessages.incrementAndGet();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    // standard constructors
}

// 消费者线程
class Consumer implements Runnable {
  
    private TransferQueue<String> transferQueue;
  
    private String name;
  
    private int numberOfMessagesToConsume;
  
    public AtomicInteger numberOfConsumedMessages
     = new AtomicInteger();
 
    @Override
    public void run() {
        for (int i = 0; i < numberOfMessagesToConsume; i++) {
            try {
                String element = transferQueue.take();
                longProcessing(element);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
 
    private void longProcessing(String element)
      throws InterruptedException {
        numberOfConsumedMessages.incrementAndGet();
        Thread.sleep(500);
    }
     
    // standard constructors
}

并行数据结构实现的类

LinkedBlockingQueue

该类实现了BlockingQueue接口，提供了一些阻塞方法，且允许限定队列中能保存的元素个数。它也实现了 Queue, Collection 和 Iterable 接口。

ConcurrentLinkedQueue

该方法实现了 Queue 接口并提供了线程安全的不限元素个数的队列。它使用一个非阻塞算法确保程序不会出现数据竞赛

LinkedBlockingDeque

该类实现了 BlockingDeque 接口，提供了具有阻塞方法的deque，且允许限制队列里元素的个数。它比LinkedBlockingQueue提供了更多的方法，但同时也具有更高的延迟，因此该类只适合当你需要 deque 性能时使用。

ConcurrentLinkedDeque

该类实现了 Deque 接口，并提供了线程安全的操作，允许你往队列的两头添加或删除数据。同样的，它比 ConcurrentLinkedQueue 具有更多的方法，却也具有更高的延迟。

ArrayBlockingQueue

该类实现了BlockingQueue接口，提供了基于数组的具有有限元素个数的blocking queue的实现。同时它实现了Queue, Collection和 Iterable 接口。和基于数组的非并行数据结构 (如 ArrayList 和 ArrayDeque) 相比，ArrayBlockingQueue在初始化时分配了数组的大小且从不改变大小。

DelayQueue

该队列里的元素必须实现 Delayed 接口，所以元素必须实现 getDelay() 方法。如果方法返回一个负数或零，说明延迟到期，队列的元素可以被取出。队列头部的第一个元素是拥有最大负延迟数的那个元素

LinkedTransferQueue

该类实现了 TransferQueue 接口。它可以用作生产者和消费者的通信通道，通过它，生产者可以等待消费者处理队列中的数据

PriorityBlockingQueue

该列中的对象根据一定的规则排序

ConcurrentHashMap

该类提供了一个线程安全的哈希表。除了 Map 接口定义方法外，该类还提供了以下方法：

search(), searchEntries(), searchKeys() 和 searchValues()：该方法允许你对哈希表中的键值对，主键或值调用搜索方法。搜索方法可以是 lambda 表达式，当搜索方法返回一个非空值时，搜索结束。该非空值即是搜索方法的输出。
reduce, reduceEntries(), reduceKeys() 和 reduceValues()：这些方法允许你使用对键值对，entry, keys 和 values做 reduce() 操作

// 以下输出结果为 9
ConcurrentHashMap<String, Integer> reducedMap = new ConcurrentHashMap<>();
        reducedMap.put("One", 1);
        reducedMap.put("Two", 2);
        reducedMap.put("Three", 3);
        System.out.println("reduce example => "
                +reducedMap.reduce(2, (k, v) -> v + 1, (total, elem) -> total + elem));

使用新特性

以下我们来看看 Java 8 引入的并行数据结构的新特性

ConcurrentHashMap的第一个例子

forEach() 方法

search() 方法

该方法接收两个参数：parallelismThreashold (如果map里的元素比指定的元素多，那么该方法将使用并行处理)；searchFunction：该搜索方法是一个 BiFunction 接口的实现

// 以下代码搜索第一本含有词语”java“的书
ExtendedProduct firstProduct=productsByBuyer.search(100,
           (id, products) -> {
               for (ExtendedProduct product: products) {
                   if (product.getTitle()
                     .toLowerCase().contains("java")) {
                       return product;
} }
           return null;
       });
       if (firstProduct!=null) {
           System.out.println(firstProduct.getBuyer()+":"+
             firstProduct.getTitle());
       }

另外还有其它的一些方法：

searchEntries(parallelismThreshold, searchFunction)
searchKeys(parallelismThreashold, searchFunction)
searchValues(parallelismThreashold, searchFunction)

reduce() 方法

该方法类似于 Stream 框架的 reduce() 方法，它接收三个参数：

parallelismThreshold：如果 ConcurrentHashMap 里的元素比指定的这个参数还多，那么这个方法就使用并行处理。
transformer：该参数是一个 lambda 函数表示的 BiFunction 接口，它接收一对键名和对应的值，然后返回对键值转换后的对象。
reducer：该参数也是一个 lambda 函数表示的 BiFunction 接口，它接收两个 transformer 方法的运行结果作为参数。该参数的目的是把两个参数合并处理成一个参数返回

compute() 方法

该方法接收两个参数，一个参数是元素的 key, 另一个参数是一个 lambda 函数表示的 BiFunction 接口。该lambda方法接收元素的key 和 value (如果key不存在则接收的value 参数为 null)。如果key存在，该方法会用lambda函数的运行结果替换掉原来的值，如果key不存在，则插入lambda方法的运行结果。如果lambda方法的运行结果为null，则从map里删除该key。

请注意在 BiFunction 运行中，一个或多个 map entries 可能被上锁。因此你的 BiFunction 不应该运行太长，也不该更改任何其他的entries。否则死锁就会产生。

merge() 方法

merge() 方法允许你把一对键值合并入 map 中。如果 key 不存在，则插入指定的值。如果 key 存在，那么最后一个 lambda 方法的运行结果将成为新的值。该方法接收三个参数：

key
value
用 lambda 表达式表示的 BiFunction 接口实现。

ConcurrentLinkedDeque类

removeIf() 方法

该方法接收一个返回值为 true 或 false 的 Predicate 接口的实现作为参数，每一个元素将调用参数里的方法，如果参数里的方法对一个元素的运行结果为 true，那么该元素将被删除。例如：

// 删除salesrank这个属性值高于 1000 的所有的产品
System.out.println("Products: "+productList.size());
       productList.removeIf(product -> product.getSalesrank() >
         1000);
       System.out.println("Products; "+productList.size());
       productList.forEach(product -> {
           System.out.println(product.getTitle()+":
             "+product.getSalesrank());
});

Atomic 变量

Atomic 变量提供了针对 integer, long, boolean, reference 和 Array 对象的原子操作。他们提供了一些方法来增加值，减少值，设值，返回值或是如果当前值和预定义的值相等时则设置新值。

Java 8 提供了四个新的类。分别是 DoubleAccumulator, DoubleAdder, LongAccumulator 和 LongAdder。LongAdder 提供了和 AtomicLong 类相似的功能，但是当需要从不同线程频繁更改这个数却只需要在操作最后访问该值时，LongAdder 具有更好的性能。

LongAccumulator 和 DoubleAccumulator 类非常相似，它们的构造函数都接收两个参数：

内部计数器的初始值
用来累加新值的方法

LongAccumulator accumulator=new LongAccumulator((x,y)
                 -> x*y, 1);
IntStream.range(1, 10).parallel().forEach(x ->
             accumulator.accumulate(x));
System.out.println(accumulator.get());

这里总结一下几个用作同步的类

公共类供以下例子使用

public class CommonTask {
    public static void doTask() {
        long duration = ThreadLocalRandom.current().nextLong(10);
        System.out.printf("%s-%s: Working %d seconds\n", new
                Date(), Thread.currentThread().getName(), duration);
        try {
            TimeUnit.SECONDS.sleep(duration);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Lock 接口

public class LockTask implements Runnable {
    private static ReentrantLock lock = new ReentrantLock();
    private String name;

    public LockTask(String name) {
        this.name = name;
    }

    @Override
    public void run() {
        try {
            lock.lock();
            System.out.println("Task: " + name + "; Date: " + new
                    Date() + ": Running the task");
            CommonTask.doTask();
            System.out.println("Task: " + name + "; Date: " + new
                    Date() + ": The execution has finished");
        } finally {
            lock.unlock();
        }
    }
}

public class LockMain {
    public static void main(String[] args) {
        ThreadPoolExecutor executor=(ThreadPoolExecutor)
                Executors.newCachedThreadPool();
        for (int i=0; i<10; i++) {
            executor.execute(new LockTask("Task "+i));
        }
        executor.shutdown();
        try {
            executor.awaitTermination(1, TimeUnit.DAYS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    } 
}

Semaphore类

public class SemaphoreTask implements Runnable{
    private Semaphore semaphore;
    public SemaphoreTask(Semaphore semaphore) {
        this.semaphore=semaphore;
    }
    @Override
    public void run() {
        try {
            semaphore.acquire();
            CommonTask.doTask();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release();
        }
    }
}

public static void main(String[] args) {
    Semaphore semaphore=new Semaphore(2);
    ThreadPoolExecutor executor=(ThreadPoolExecutor)
            Executors.newCachedThreadPool();
    for (int i=0; i<10; i++) {
        executor.execute(new SemaphoreTask(semaphore));
    }
    executor.shutdown();
    try {
        executor.awaitTermination(1, TimeUnit.DAYS);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

CountDownLatch 类

public class CountDownTask implements Runnable {
    private CountDownLatch countDownLatch;
    public CountDownTask(CountDownLatch countDownLatch) {
        this.countDownLatch=countDownLatch;
    }
    @Override
    public void run() {
        CommonTask.doTask();
        countDownLatch.countDown();
    }
}

public static void main(String[] args) {
    CountDownLatch countDownLatch=new CountDownLatch(10);
    ThreadPoolExecutor executor=(ThreadPoolExecutor)
            Executors.newCachedThreadPool();
    System.out.println("Main: Launching tasks");
    for (int i=0; i<10; i++) {
        executor.execute(new CountDownTask(countDownLatch));
    }
    try {
        countDownLatch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    executor.shutdown();
}

CyclicBarrier 类

public class BarrierTask implements Runnable {
    private CyclicBarrier barrier;
    public BarrierTask(CyclicBarrier barrier) {
        this.barrier=barrier;
    }
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName()+":Phase 1");
        CommonTask.doTask();
        try {
            barrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()+":Phase 2");
    }
}

public class FinishBarrierTask implements Runnable {
    @Override
    public void run() {
        System.out.println("FinishBarrierTask: All the tasks have finished");
    }
}

public static void main(String[] args) {
    CyclicBarrier barrier=new CyclicBarrier(10,new
            FinishBarrierTask());
    ThreadPoolExecutor executor=(ThreadPoolExecutor)
            Executors.newCachedThreadPool();
    for (int i=0; i<10; i++) {
        executor.execute(new BarrierTask(barrier));
    }
    executor.shutdown();
    try {
        executor.awaitTermination(1, TimeUnit.DAYS);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

CompletableFuture 类

CompletableFuture允许你实现和任务相关的事件驱动模型。CompletableFuture代表了一个异步计算的结果，但是其结果可以被任何一个线程所创建。可以通过它的 complete() 方法在当计算顺利完成时创建运算结果，或使用 completeExceptionally() 方法当计算抛出异常时。如果有两个线程调用同一个 CompletableFuture 的 complete() 或 completeExceptionally()，只有第一个调用有效。

你可以使用构造函数创建一个 CompletableFuture，然后使用 complete() 方法获取最终结果。你也可以使用 runAsync() 或 supplyAsync() 方法创建 CompletableFuture。runAsync() 方法运行一个 Runnable 对象并返回 CompletableFuture<Void> 所以你的运行结果不能有返回值。supplyAsync() 方法运行一个 Supplier接口实现。它可以有返回值。CompletableFuture类提供了很多方法允许你组织任务的运行顺序，这些方法包括：

thenApplyAsync()：该方法接收一个用 lambda 函数表达的 Function 接口实现作为参数，参数中的 lambda 函数在 CompletableFuture 调用结束后运行。该方法返回另一个 CompletableFuture 用以获取函数的运行结果
thenComposeAsync()：该方法和 thenApplyAsync() 方法类似。但是在参数中的 lambda 表达式返回的值也是 CompletableFuture 时非常有用。

// 假设有如下两个方法 getUserInfo 和 getUserRating
// 它们都返回一个 CompletableFuture
public CompletableFuture<UserInfo> userInfo = getUserInfo(userId)
public CompletableFuture<UserRating> getUserRating(UserInfo)

// 如果使用thenApplyAsync，返回结果是这样
CompletableFuture<CompletableFuture<UserRating>> f =
    userInfo.thenApplyAsync(this::getUserRating);

// 但是如果使用 thenComposeAsync，返回结果则简单很多
// thenComposeAsync 有点像flatMap() 方法，能够把元素展开来
CompletableFuture<UserRating> relevanceFuture =
    userInfo.thenComposeAsync(this::getUserRating);

thenAcceptAsync()：该方法的参数是一个用 lambda 函数表达的 Consumer 接口实现，因此它没有返回值
thenRunAsync()：该方法和上一个类似，但是参数里函数是一个 Runnable 对象
thenCombineAsync()：该方法接收两个参数，第一个参数为另一个 CompletableFuture 实例，第二个参数为一个 BiFunction 接口实现。当两个 CompletableFuture 运行结束后，BiFunction 会被执行。该方法返回一个 CompletableFuture 用来获取 BiFunction 的运行结果
runAfterBothAsync()：和上一个方法类似，但是第二个参数是一个 Runnable 对象，该 Runnable 在两个 CompletableFuture 运行结束后被调用
runAfterEitherAsync()：该方法和上一个方法类似，但是第二个参数 Runnable 任务在两个 CompletableFuture 的其中一个运行结束后被调用
allOf()：该方法接收多个 CompletableFuture 对象。当所有的CompletableFuture对象运行结束后它返回一个 CompletableFuture<Void> 对象
anyOf()：该方法接收多个 CompletableFuture 对象。当任何一个 CompletableFuture 对象运行结束后它返回运行结果的 CompletableFuture 对象

最后你通过 get() 或 join() 方法获取 CompletableFuture 的运行结果。两个方法都阻塞了调用线程直到 CompletableFuture 运行结束并返回结果。两个方法的不同之处在于 get() 方法抛出 ExecutionException (checked exception)， join() 方法则抛出RuntimeException (unchecked exception)。因此在不抛异常的 lambdas (如 Supplier, Consumer 或 Runnable 里) 使用 join() 更为方便。

那些方法名以 Async 结尾的意味着这些方法将使用 ForkJoinPool.commonPool 实例并行运行。方法名没有 Async 的则使用串行方法运行。

以下是如何使用CompletableFuture的代码示例：

public class LoadTask implements Supplier<List<Product>> {
    private Path path;
    public LoadTask (Path path) {
        this.path = path;
    }

    @Override
    public List<Product> get() {
        List<Product> productList=null;
        try {
            productList = Files.walk(path, FileVisitOption.FOLLOW_LINKS)
                    .parallel()
                    .filter(f -> f.toString().endsWith(".txt"))
                    .map(ProductLoader::load).collect
                            (Collectors.toList());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return productList;
    }
}

public class SearchTask implements Function<List<Product>,
        List<Product>> {
    private String query;
    public SearchTask(String query) {
        this.query=query;
    }
    @Override
    public List<Product> apply(List<Product> products) {
        System.out.println(new Date()+": CompletableTask: start");
        List<Product> ret = products.stream()
                .filter(product -> product.getTitle()
                        .toLowerCase().contains(query))
                .collect(Collectors.toList());
        System.out.println(new Date()+": CompletableTask: end:"+ret.size());
        return ret;
    }
}

public class WriteTask implements Consumer<List<Product>> {
    @Override
    public void accept(List<Product> products) {
        // implementation is omitted
    }
}

public class CompletableMain {
    public static void main(String[] args) {
        Path file = Paths.get("data","category");
        System.out.println(new Date() + ": Main: Loading products");
        LoadTask loadTask = new LoadTask(file);
        CompletableFuture<List<Product>> loadFuture =
                CompletableFuture.supplyAsync(loadTask);
        System.out.println(new Date() + ": Main: Then apply for search");
        CompletableFuture<List<Product>> completableSearch =
                loadFuture.thenApplyAsync(new SearchTask("love"));
        CompletableFuture<Void> completableWrite =
                completableSearch.thenAcceptAsync(new WriteTask());
        completableWrite.exceptionally(ex -> {
            System.out.println(new Date() + ": Main: Exception " + ex.getMessage());
            return null;
        });

        System.out.println(new Date() + ": Main: Then apply for users");
        CompletableFuture<List<String>> completableUsers =
                loadFuture.thenApplyAsync(resultList -> {
                                System.out.println(new Date() + ": Main: Completable users: start");
                                List<String> users =
                                        resultList.stream()
                                                .flatMap(p -> p.getReviews().stream())
                                                .map(review -> review.getUser())
                                                .distinct()
                                                .collect(Collectors.toList());
                                System.out.println(new Date() + ": Main: Completable users: end");
                                return users;
                            });
        System.out.println(new Date() + ": Main: Then apply for best rated product....");
        CompletableFuture<Product> completableProduct = loadFuture
                .thenApplyAsync(resultList -> {
                    Product maxProduct = null;
                    double maxScore = 0.0;
                    System.out.println(new Date() + ": Main: Completable product: start");
                    for (Product product : resultList) {
                        if (!product.getReviews().isEmpty()) {
                            double score = product.getReviews().stream()
                                                    .mapToDouble(review ->
                                                            review.getValue())
                                                    .average().getAsDouble();
                            if (score > maxScore) {
                                maxProduct = product;
                                maxScore = score;
                            } 
                        }
                    }
                    System.out.println(new Date() + ": Main: Completable product: end");
                    return maxProduct;
                });
        System.out.println(new Date() + ": Main: Then apply for best selling product....");
        CompletableFuture<Product> completableBestSellingProduct =
                loadFuture
                        .thenApplyAsync(resultList -> {
                            System.out.println(new Date() + ": Main: Completable best selling: start");
                            Product bestProduct = resultList
                                    .stream()
                                    .min(Comparator.comparingLong
                                            (Product::getSalesrank))
                                    .orElse(null);
                            System.out.println(new Date() + ": Main: Completable best selling: end");
                            return bestProduct;
                        });
        CompletableFuture<String> completableProductResult =
                completableBestSellingProduct
                        .thenCombineAsync(
                                completableProduct,
                                (bestSellingProduct, bestRatedProduct) -> {
                                    System.out.println(new Date() + ": Main: Completable product result: start");
                                    String ret = "The best selling product is "
                                            + bestSellingProduct.getTitle() + "\n";
                                    ret += "The best rated product is "
                                            + bestRatedProduct.getTitle();
                                    System.out.println(new Date() + ": Main: Completable product result: end");
                                    return ret;
                                });
        System.out.println(new Date() + ": Main: Waiting for results");
                CompletableFuture<Void> finalCompletableFuture =
                        CompletableFuture
                                .allOf(completableProductResult, completableUsers,
                                        completableWrite);
        finalCompletableFuture.join();
        try {
            System.out.println("Number of loaded products: " + loadFuture.get().size());
            System.out.println("Number of found products: " + completableSearch.get().size());
            System.out.println("Number of users: " + completableUsers.get().size());
            System.out.println("Best rated product: " + completableProduct.get().getTitle());
            System.out.println("Best selling product: " + completableBestSellingProduct.get().getTitle());
            System.out.println("Product result: "+completableProductResult.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

分享到：

第八章 - Map与Collect模型

2019-09-24 13:31
浏览 411
评论(0)
分类:编程语言
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客 更多访客>>