lambda 高并發

因此Java 8不久前發布，具有大量功能和更改。 我們所有的Java狂熱者一直在等待這一點，從他們最初宣布Java 7的所有強大功能開始一直到最終被取消。

我最近才有時間實際開始給它一個真實的外觀，我將我的家庭項目更新為8個，我不得不說，我對所獲得的一切通常感到非常滿意。 java.time API的“模仿” JodaTime是一個很大的改進，java.util.stream軟件包正在變得有用，lambda將改變我們的編碼樣式，這可能需要一些時間來適應這些變化……引用“強大的力量伴隨著巨大的責任”這句話是正確的，我認為未來可能會有一些有趣的時刻，因為編寫一些難以破解的代碼非常容易。 作為調試我下面編寫的代碼的示例，它會很有趣。

該文件示例在我的Github博客倉庫中

此示例的操作很簡單，運行幾個線程，并發執行一些工作，然后等待它們全部完成。 我在玩Java 8的時候就想通了，讓我全力以赴……
這是我想出的：

package net.briandupreez.blog.java8.futures;import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory;import java.util.Collection; import java.util.List; import java.util.concurrent.*; import java.util.stream.Collectors;/*** Generified future running and completion** @param <T> the result type* @param <S> the task input*/ public class WaitingFuturesRunner<T, S> {private transient static final Log logger = LogFactory.getLog(WaitingFuturesRunner.class);private final Collection<Task<T, S>> tasks;private final long timeOut;private final TimeUnit timeUnit;private final ExecutorService executor;/*** Constructor, used to initialise with the required tasks** @param tasks the list of tasks to execute* @param timeOut max length of time to wait* @param timeUnit time out timeUnit*/public WaitingFuturesRunner(final Collection<Task<T, S>> tasks, final long timeOut, final TimeUnit timeUnit) {this.tasks = tasks;this.timeOut = timeOut;this.timeUnit = timeUnit;this.executor = Executors.newFixedThreadPool(tasks.size());}/*** Go!** @param taskInput The input to the task* @param consolidatedResult a container of all the completed results*/public void go(final S taskInput, final ConsolidatedResult<T> consolidatedResult) {final CountDownLatch latch = new CountDownLatch(tasks.size());final List<CompletableFuture<T>> theFutures = tasks.stream().map(aSearch -> CompletableFuture.supplyAsync(() -> processTask(aSearch, taskInput, latch), executor)).collect(Collectors.<CompletableFuture<T>>toList());final CompletableFuture<List<T>> allDone = collectTasks(theFutures);try {latch.await(timeOut, timeUnit);logger.debug("complete... adding results");allDone.get().forEach(consolidatedResult::addResult);} catch (final InterruptedException | ExecutionException e) {logger.error("Thread Error", e);throw new RuntimeException("Thread Error, could not complete processing", e);}}private <E> CompletableFuture<List<E>> collectTasks(final List<CompletableFuture<E>> futures) {final CompletableFuture<Void> allDoneFuture = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));return allDoneFuture.thenApply(v -> futures.stream().map(CompletableFuture<E>::join).collect(Collectors.<E>toList()));}private T processTask(final Task<T, S> task, final S searchTerm, final CountDownLatch latch) {logger.debug("Starting: " + task);T searchResults = null;try {searchResults = task.process(searchTerm, latch);} catch (final Exception e) {e.printStackTrace();}return searchResults;}}

測試：

package net.briandupreez.blog.java8.futures;import net.briandupreez.blog.java8.futures.example.StringInputTask; import net.briandupreez.blog.java8.futures.example.StringResults; import org.apache.log4j.BasicConfigurator; import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test;import java.util.ArrayList; import java.util.List; import java.util.concurrent.TimeUnit;/*** Test* Created by brian on 4/26/14.*/ public class CompletableFuturesRunnerTest {@BeforeClasspublic static void init() {BasicConfigurator.configure();}/*** 5tasks at 3000ms concurrently should not be more than 3100* @throws Exception error*/@Test(timeout = 3100)public void testGo() throws Exception {final List<Task<String, String>> taskList = setupTasks();final WaitingFuturesRunner<String, String> completableFuturesRunner = new WaitingFuturesRunner<>(taskList, 4, TimeUnit.SECONDS);final StringResults consolidatedResults = new StringResults();completableFuturesRunner.go("Something To Process", consolidatedResults);Assert.assertEquals(5, consolidatedResults.getResults().size());for (final String s : consolidatedResults.getResults()) {Assert.assertTrue(s.contains("complete"));Assert.assertTrue(s.contains("Something To Process"));}}private List<Task<String, String>> setupTasks() {final List<Task<String, String>> taskList = new ArrayList<>();final StringInputTask stringInputTask = new StringInputTask("Task 1");final StringInputTask stringInputTask2 = new StringInputTask("Task 2");final StringInputTask stringInputTask3 = new StringInputTask("Task 3");final StringInputTask stringInputTask4 = new StringInputTask("Task 4");final StringInputTask stringInputTask5 = new StringInputTask("Task 5");taskList.add(stringInputTask);taskList.add(stringInputTask2);taskList.add(stringInputTask3);taskList.add(stringInputTask4);taskList.add(stringInputTask5);return taskList;} }

輸出：

0 [pool-1-thread-1] Starting: StringInputTask{taskName='Task 1'}0 [pool-1-thread-5] Starting: StringInputTask{taskName='Task 5'}0 [pool-1-thread-2] Starting: StringInputTask{taskName='Task 2'}2 [pool-1-thread-4] Starting: StringInputTask{taskName='Task 4'}2 [pool-1-thread-3] Starting: StringInputTask{taskName='Task 3'}3003 [pool-1-thread-5] Done: Task 53004 [pool-1-thread-3] Done: Task 33003 [pool-1-thread-1] Done: Task 13003 [pool-1-thread-4] Done: Task 43003 [pool-1-thread-2] Done: Task 23007 [Thread-0] WaitingFuturesRunner ?- complete... adding results

在執行此操作時，我發現并閱讀了一些有用的文章/鏈接：

Oracle： Lambda教程

IBM： Java 8并發

Tomasz Nurkiewicz： CompletableFuture權威指南

翻譯自: https://www.javacodegeeks.com/2014/04/playing-with-java-8-lambdas-and-concurrency.html

lambda 高并發

總結

以上是生活随笔為你收集整理的lambda 高并发_玩Java 8 – Lambda和并发的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。