lambda 高并發(fā)

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

我最近才有時(shí)間實(shí)際開(kāi)始給它一個(gè)真實(shí)的外觀(guān)，我將我的家庭項(xiàng)目更新為8個(gè)，我不得不說(shuō)，我對(duì)所獲得的一切通常感到非常滿(mǎn)意。 java.time API的“模仿” JodaTime是一個(gè)很大的改進(jìn)，java.util.stream軟件包正在變得有用，lambda將改變我們的編碼樣式，這可能需要一些時(shí)間來(lái)適應(yīng)這些變化……引用“強(qiáng)大的力量伴隨著巨大的責(zé)任”這句話(huà)是正確的，我認(rèn)為未來(lái)可能會(huì)有一些有趣的時(shí)刻，因?yàn)榫帉?xiě)一些難以破解的代碼非常容易。 作為調(diào)試我下面編寫(xiě)的代碼的示例，它會(huì)很有趣。

該文件示例在我的Github博客倉(cāng)庫(kù)中

此示例的操作很簡(jiǎn)單，運(yùn)行幾個(gè)線(xiàn)程，并發(fā)執(zhí)行一些工作，然后等待它們?nèi)客瓿伞?我在玩Java 8的時(shí)候就想通了，讓我全力以赴……
這是我想出的：

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;}}

測(cè)試：

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

在執(zhí)行此操作時(shí)，我發(fā)現(xiàn)并閱讀了一些有用的文章/鏈接：

Oracle： Lambda教程

IBM： Java 8并發(fā)

Tomasz Nurkiewicz： CompletableFuture權(quán)威指南

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

lambda 高并發(fā)

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。