Tùy chọn không chặn Akka khi phản hồi HTTP được yêu cầu

Question

Tôi hiểu cách tạo ứng dụng không chặn thư trong akka và có thể dễ dàng giả lập các ví dụ thực hiện hoạt động đồng thời và trả lại kết quả tổng hợp trong thư. Tôi gặp khó khăn khi hiểu những gì các tùy chọn không chặn của tôi là khi ứng dụng của tôi phải trả lời yêu cầu HTTP. Mục tiêu là để nhận được yêu cầu và ngay lập tức giao nó cho một diễn viên địa phương hoặc từ xa để thực hiện công việc, do đó sẽ tắt nó để có kết quả là có thể mất một thời gian. Thật không may theo mô hình này, tôi không hiểu làm thế nào tôi có thể thể hiện điều này với một loạt không bị chặn "nói" thay vì chặn "yêu cầu". Nếu tại bất kỳ điểm nào trong chuỗi tôi sử dụng, tôi không còn có tương lai là sử dụng làm nội dung phản hồi cuối cùng (yêu cầu bởi giao diện khung http trong trường hợp này là finagle - nhưng đó không phải là quan trọng). Tôi hiểu rằng yêu cầu là một chủ đề riêng và ví dụ của tôi khá giả tạo nhưng chỉ cố gắng hiểu các tùy chọn thiết kế của tôi.

Tóm lại, Nếu ví dụ giả tạo của tôi dưới đây có thể được làm lại để chặn ít hơn, tôi rất thích hiểu làm thế nào. Đây là lần đầu tiên sử dụng akka kể từ khi khám phá ánh sáng một năm + trước và trong mỗi bài viết, tài liệu và nói chuyện tôi đã xem nói không chặn các dịch vụ.

Câu trả lời khái niệm có thể hữu ích nhưng cũng có thể giống như những gì tôi đã đọc. Làm việc/chỉnh sửa ví dụ của tôi có thể là chìa khóa cho sự hiểu biết của tôi về vấn đề chính xác mà tôi đang cố giải quyết. Nếu ví dụ hiện tại thường là những gì cần phải được thực hiện thì xác nhận đó cũng hữu ích, vì vậy tôi không tìm kiếm ma thuật không tồn tại.

Lưu ý Các bí danh sau đây:. Nhập khẩu com.twitter.util {tương lai => TwitterFuture, chờ đợi => TwitterAwait}

object Server { 

    val system = ActorSystem("Example-System") 

    implicit val timeout = Timeout(1 seconds) 

    implicit def scalaFuture2twitterFuture[T](scFuture: Future[T]): TwitterFuture[T] = { 
    val promise = TwitterPromise[T] 
    scFuture onComplete { 
     case Success(result) ⇒ promise.setValue(result) 
     case Failure(failure) ⇒ promise.setException(failure) 
    } 
    promise 
    } 

    val service = new Service[HttpRequest, HttpResponse] { 
    def apply(req: HttpRequest): TwitterFuture[HttpResponse] = req.getUri match { 
     case "https://stackoverflow.com/a/b/c" => 
     val w1 = system.actorOf(Props(new Worker1)) 

     val r = w1 ? "take work" 

     val response: Future[HttpResponse] = r.mapTo[String].map { c => 
      val resp = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK) 
      resp.setContent(ChannelBuffers.copiedBuffer(c, CharsetUtil.UTF_8)) 
      resp 
     } 

     response 
    } 
    } 

//val server = Http.serve(":8080", service); TwitterAwait.ready(server) 

    class Worker1 extends Actor with ActorLogging { 
    def receive = { 
     case "take work" => 
     val w2 = context.actorOf(Props(new Worker2)) 
     pipe (w2 ? "do work") to sender 
    } 
    } 

    class Worker2 extends Actor with ActorLogging { 
    def receive = { 
     case "do work" => 
     //Long operation... 
     sender ! "The Work" 
    } 
    } 

    def main(args: Array[String]) { 
    val r = service.apply(
     com.twitter.finagle.http.Request("https://stackoverflow.com/a/b/c") 
    ) 
    println(TwitterAwait.result(r).getContent.toString(CharsetUtil.UTF_8)) // prints The Work 
    } 
} 

Cảm ơn trước cho bất kỳ hướng dẫn được cung cấp!

Answer 1

Bạn có thể tránh việc gửi một tương lai như một thông điệp bằng pipe pattern -ie, trong Worker1 bạn muốn viết:

pipe(w2 ? "do work") to sender 

Thay vì:

sender ! (w2 ? "do work") 

Bây giờ r sẽ là một Future[String] thay vì một số Future[Future[String]].

---

Cập nhật: giải pháp pipe ở trên là cách chung để tránh diễn viên của bạn phản hồi trong tương lai. Như Viktor chỉ ra trong một chú thích dưới đây, trong trường hợp này bạn có thể mất bạn Worker1 ra khỏi vòng lặp hoàn toàn bằng cách nói Worker2 để trả lời trực tiếp các diễn viên mà nó (Worker1) nhận được thông báo từ:

w2.tell("do work", sender) 

won này Không phải là tùy chọn nếu Worker1 chịu trách nhiệm điều hành phản hồi từ Worker2 theo cách nào đó (bằng cách sử dụng map trên w2 ? "do work", kết hợp nhiều tương lai với flatMap hoặc for-comprehrehension, v.v.), nhưng nếu không cần thiết, phiên bản này là sạch hơn và hiệu quả hơn.

---

Điều đó giết chết một Await.result. Bạn có thể thoát khỏi người kia bằng cách viết một cái gì đó như sau:

val response: Future[HttpResponse] = r.mapTo[String].map { c => 
    val resp = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK) 
    resp.setContent(ChannelBuffers.copiedBuffer(c, CharsetUtil.UTF_8)) 
    resp 
} 

Bây giờ bạn chỉ cần bật Future này thành một TwitterFuture. Tôi không thể nói với bạn khỏi đầu của tôi chính xác làm thế nào để làm điều này, nhưng nó phải là fairly trivial, và chắc chắn không yêu cầu chặn.

Answer 2

Bạn chắc chắn không phải chặn tất cả ở đây. Thứ nhất, cập nhật nhập khẩu của bạn cho những thứ twitter để:

import com.twitter.util.{Future => TwitterFuture, Await => TwitterAwait, Promise => TwitterPromise} 

Bạn sẽ cần twitter Promise như đó là impl của Future bạn sẽ trở lại từ phương pháp apply. Sau đó, hãy làm theo những gì Travis Brown nói trong câu trả lời của mình để diễn viên của bạn phản ứng theo cách mà bạn không có tương lai lồng nhau. Một khi bạn làm điều đó, bạn sẽ có thể thay đổi phương thức apply của bạn để một cái gì đó như thế này:

def apply(req: HttpRequest): TwitterFuture[HttpResponse] = req.getUri match { 
    case "https://stackoverflow.com/a/b/c" => 
    val w1 = system.actorOf(Props(new Worker1)) 

    val r = (w1 ? "take work").mapTo[String] 
    val prom = new TwitterPromise[HttpResponse] 
    r.map(toResponse) onComplete{ 
     case Success(resp) => prom.setValue(resp) 
     case Failure(ex) => prom.setException(ex)    
    } 

    prom 
} 

def toResponse(c:String):HttpResponse = { 
    val resp = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK) 
    resp.setContent(ChannelBuffers.copiedBuffer(c, CharsetUtil.UTF_8)) 
    resp 
} 

này có lẽ cần có một công việc ít hơn. Tôi đã không thiết lập nó trong IDE của tôi, vì vậy tôi không thể đảm bảo bạn nó biên dịch, nhưng tôi tin rằng ý tưởng là âm thanh. Những gì bạn trở về từ phương pháp apply là TwitterFuture chưa được hoàn thành. Nó sẽ được hoàn thành khi tương lai từ diễn viên hỏi (?) Được thực hiện và đó là hạnh phúc thông qua một không bị chặn onComplete gọi lại.