标签: future

我是scala期货的新手,我对scala期货的回报价值有疑问.

因此,scala未来的语法通常是

 def downloadPage(url: URL) = Future[List[Int]] {

 }

我想知道如何访问List[Int]调用此方法的其他方法.

换一种说法,

val result = downloadPage("localhost") 

那么List[Int]走出未来的方法应该是什么？

我尝试过使用map方法,但无法成功完成此操作

我应该什么时候使用std::promise过std::async或std::packaged_task？你能告诉我何时使用它们的实际例子吗？

在异步JavaScript中,很容易并行运行任务并等待所有这些任务完成Promise.all:

async function bar(i) {
  console.log('started', i);
  await delay(1000);
  console.log('finished', i);
}

async function foo() {
    await Promise.all([bar(1), bar(2)]);
}

// This works too:
async function my_all(promises) {
    for (let p of promises) await p;
}

async function foo() {
    await my_all([bar(1), bar(2), bar(3)]);
}

我试图在python中重写后者:

import asyncio

async def bar(i):
  print('started', i)
  await asyncio.sleep(1)
  print('finished', i)

async def aio_all(seq):
  for f in seq:
    await f

async def main():
  await aio_all([bar(i) for i in range(10)])

loop = asyncio.get_event_loop() …

我有这个问题,我每次都要解决这个问题.我无法使用for comprehension来映射Future中包含的内容.

例:

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

val f = Future( List("A", "B", "C") )
for {
  list <- f
  e <- list
} yield (e -> 1)

这给了我错误:

 error: type mismatch;
 found   : List[(String, Int)]
 required: scala.concurrent.Future[?]
              e <- list
                ^

但是,如果我这样做,它工作正常:

f.map( _.map( (_ -> 1) ) )

如果我不能通过使用for comprehension来做到这一点,那么它在我的另一个例子中是否适用于我不进行flatmap的原因？我正在使用Scala 2.10.0.

在concurrent.futures.Executor.map采用可变数目从该给出的函数被调用iterables的.如果我有一个生成通常解压缩的元组的生成器,我应该如何调用它？

以下方法不起作用,因为每个生成的元组都作为map的不同参数给出:

args = ((a, b) for (a, b) in c)
for result in executor.map(f, *args):
    pass

如果没有生成器,映射的所需参数可能如下所示:

executor.map(
    f,
    (i[0] for i in args),
    (i[1] for i in args),
    ...,
    (i[N] for i in args),
)

我有一个方法,通过超时执行一些任务.我使用ExecutorServer.submit()来获取Future对象,然后使用超时调用future.get().这工作正常,但我的问题是处理我的任务可以抛出的已检查异常的最佳方法.以下代码有效,并保留已检查的异常,但如果方法签名中的已检查异常列表发生更改,则它似乎非常笨拙且容易中断.

对于如何解决这个问题,有任何的建议吗？我需要针对Java 5,但我也很想知道在较新版本的Java中是否有好的解决方案.

public static byte[] doSomethingWithTimeout( int timeout ) throws ProcessExecutionException, InterruptedException, IOException, TimeoutException {

    Callable<byte[]> callable = new Callable<byte[]>() {
        public byte[] call() throws IOException, InterruptedException, ProcessExecutionException {
            //Do some work that could throw one of these exceptions
            return null;
        }
    };

    try {
        ExecutorService service = Executors.newSingleThreadExecutor();
        try {
            Future<byte[]> future = service.submit( callable );
            return future.get( timeout, TimeUnit.MILLISECONDS );
        } finally {
            service.shutdown();
        }
    } catch( Throwable t ) { //Exception handling of nested exceptions is …

我只是在探索java.util.concurrent包.

我了解到类' Future '有一个方法boolean cancel(boolean mayInterruptIfRunning)

请查找附上我写的测试代码:

package com.java.util.concurrent;

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;

public class FutureTester {

/**
 * @param args
 * @throws InterruptedException
 */
public static void main(String[] args) throws InterruptedException {
    // TODO Auto-generated method stub
    int poolCnt = 1;
    Callable<NumberPrinter> numberPrinter = null;
    ScheduledThreadPoolExecutor schPool = new ScheduledThreadPoolExecutor(
            poolCnt);
    ScheduledFuture<NumberPrinter>[] numPrinterFutures = new ScheduledFuture[poolCnt];
    FutureTask<NumberPrinter>[] futureTask = new FutureTask[poolCnt];

    for (int i = 0; i < poolCnt; i++) {
        numberPrinter = new …

介绍

Scala Future(在2.10和现在的2.9.3中是新的)是一个应用程序的函子,这意味着如果我们有一个可遍历的类型 F,我们可以使用F[A]一个函数A => Future[B]并将它们变成一个函数Future[F[B]].

此操作在标准库中可用Future.traverse.如果我们从库中导入applicative functor实例,Scalaz 7还提供了更通用的功能.traverseFuturescalaz-contrib

这两种traverse方法在流的情况下表现不同.标准库遍历在返回之前使用流,而Scalaz会立即返回未来:

import scala.concurrent._
import ExecutionContext.Implicits.global

// Hangs.
val standardRes = Future.traverse(Stream.from(1))(future(_))

// Returns immediately.
val scalazRes = Stream.from(1).traverse(future(_))

还有另一个不同之处,正如Leif Warner 在此观察到的那样.标准库traverse立即启动所有异步操作,而Scalaz启动第一个,等待它完成,启动第二个,等待它,依此类推.

流的不同行为

通过编写一个将为流中的第一个值休眠几秒钟的函数来显示第二个差异非常容易:

def howLong(i: Int) = if (i == 1) 10000 else 0

import scalaz._, Scalaz._ …

我正在学习在Scala中使用async/await.我在https://github.com/scala/async中看过这个

从理论上讲,这段代码是异步的(非阻塞),但它没有并行化:

def slowCalcFuture: Future[Int] = ...             
def combined: Future[Int] = async {               
   await(slowCalcFuture) + await(slowCalcFuture)
}
val x: Int = Await.result(combined, 10.seconds)    

而另一个是并行化的:

def combined: Future[Int] = async {
  val future1 = slowCalcFuture
  val future2 = slowCalcFuture
  await(future1) + await(future2)
}

它们之间的唯一区别是使用中间变量.这怎么会影响并行化？

我正在尝试使用Akka HTTP来基本验证我的请求.碰巧我有一个外部资源来进行身份验证,因此我必须对此资源进行休息调用.

这需要一些时间,并且在处理时,我的API的其余部分似乎被阻止,等待此调用.我用一个非常简单的例子重现了这个:

// used dispatcher:
implicit val system = ActorSystem()
implicit val executor = system.dispatcher
implicit val materializer = ActorMaterializer()


val routes = 
  (post & entity(as[String])) { e =>
    complete {
      Future{
        Thread.sleep(5000)
        e
      }
    }
  } ~
  (get & path(Segment)) { r =>
    complete {
      "get"
    }
  }

如果我发布到日志端点,我的get端点也会等待5秒,这是日志端点所指示的.

这是预期的行为,如果是,如何在不阻止整个API的情况下进行阻止操作？