Thực hiện các tác vụ ràng buộc CPU với các diễn viên Scala?

Question

Giả sử tôi phải thực hiện một số tác vụ bị ràng buộc CPU. Nếu tôi có 4 CPU, ví dụ, tôi có lẽ sẽ tạo ra một hồ bơi thread kích thước cố định của 4-5 chủ đề công nhân chờ đợi trên một hàng đợi và đặt các nhiệm vụ trong hàng đợi. Trong Java, tôi có thể sử dụng java.util.concurrent (có thể là ThreadPoolExecutor) để thực hiện cơ chế này.

Bạn sẽ triển khai nó như thế nào với diễn viên Scala?

Answer 1

Tất cả các diễn viên về cơ bản là các chuỗi được thực thi bởi một người lên lịch dưới mui xe. Trình lập lịch biểu tạo một nhóm luồng để thực thi các tác nhân gần như ràng buộc với số lõi của bạn. Điều này có nghĩa rằng bạn chỉ có thể tạo ra một diễn viên mỗi công việc bạn cần phải thực hiện và để lại phần còn lại cho Scala:

for(i <- 1 to 20) { 
    actor { 
    print(i); 
    Thread.sleep(1000); 
    } 
} 

Những bất lợi ở đây là tùy thuộc vào số nhiệm vụ, chi phí của việc tạo ra một chủ đề cho mỗi công việc có thể khá đắt vì các chủ đề không quá rẻ trong Java.

Một cách đơn giản để tạo ra một hồ bơi bị chặn của các diễn viên người lao động và sau đó phân phối các nhiệm vụ đối với họ thông qua tin nhắn sẽ là một cái gì đó như:

import scala.actors.Actor._ 

val numWorkers = 4 
val pool = (1 to numWorkers).map { i => 
    actor { 
    loop { 
     react { 
     case x: String => println(x) 
     } 
    } 
    } 
} 

for(i <- 1 to 20) { 
    val r = (new util.Random).nextInt(numWorkers) 
    pool(r) ! "task "+i 
} 

Lý do chúng tôi muốn tạo nhiều diễn viên là do một quá trình diễn viên duy nhất chỉ có một thông điệp (tức là nhiệm vụ) tại một thời điểm để có được tính song song cho các nhiệm vụ của bạn, bạn cần phải tạo nhiều.

Lưu ý phụ: trình lên lịch mặc định trở nên đặc biệt quan trọng khi nói đến các tác vụ ràng buộc I/O, vì bạn chắc chắn sẽ muốn thay đổi kích thước của nhóm luồng trong trường hợp đó. Hai bài đăng trên blog tốt đi sâu vào chi tiết về điều này là: Explore the Scheduling of Scala Actors và Scala actors thread pool pitfall.

Với điều đó đã nói, Akka là khung công tác cung cấp công cụ cho quy trình công việc nâng cao hơn với diễn viên và đó là những gì tôi sẽ sử dụng trong bất kỳ ứng dụng thực tế nào. Đây là một cân bằng tải (chứ không phải là ngẫu nhiên) nhiệm vụ thi hành di chúc:

import akka.actor.Actor 
import Actor._ 
import akka.routing.{LoadBalancer, CyclicIterator} 

class TaskHandler extends Actor { 
    def receive = { 
    case t: Task => 
     // some computationally expensive thing 
     t.execute 
    case _ => println("default case is required in Akka...") 
    } 
} 

class TaskRouter(numWorkers: Int) extends Actor with LoadBalancer { 
    val workerPool = Vector.fill(numWorkers)(actorOf[TaskHandler].start()) 
    val seq = new CyclicIterator(workerPool) 
} 

val router = actorOf(new TaskRouter(4)).start() 

for(i <- 1 to 20) { 
    router ! Task(..) 
} 

Bạn có thể có các loại khác nhau của Load Balancing (CyclicIterator là phân phối round-robin), vì vậy bạn có thể kiểm tra các tài liệu here để biết thêm.

Answer 2

Vâng, bạn thường không làm như vậy. Một phần của sự hấp dẫn của việc sử dụng diễn viên là họ xử lý các chi tiết đó cho bạn.

Nếu, tuy nhiên, bạn nhấn mạnh vào việc quản lý điều đó, bạn sẽ cần ghi đè phương pháp được bảo vệ scheduler trên lớp Actor của bạn để trả về IScheduler thích hợp. Xem thêm scala.actors.scheduler package và các nhận xét trên Actor trait về lịch trình liên quan.