วิธีการใหม่WorkStealingPool

ลองหา เมธอด WorkStealingPool ใหม่ซึ่งเตรียมExecutorServiceให้เรา

พูลเธรดนี้พิเศษ พฤติกรรมของมันขึ้นอยู่กับแนวคิดของการ "ขโมย" งาน

งานอยู่ในคิวและแจกจ่ายระหว่างโปรเซสเซอร์ แต่ถ้าโปรเซสเซอร์ไม่ว่าง โปรเซสเซอร์ฟรีตัวอื่นสามารถขโมยงานจากโปรเซสเซอร์นั้นและดำเนินการได้ รูปแบบนี้ถูกนำมาใช้ใน Java เพื่อลดความขัดแย้งในแอปพลิเคชันแบบมัลติเธรด มันถูกสร้างขึ้นบนเฟรมเวิร์กfork/join

ส้อม / เข้าร่วม

ใน เฟรมเวิร์ก fork/joinงานจะถูกแยกย่อยแบบเรียกซ้ำ นั่นคือ แบ่งงานออกเป็นงานย่อย จากนั้นงานย่อยจะถูกดำเนินการแยกกัน และผลลัพธ์ของงานย่อยจะถูกรวมเข้าด้วยกันเพื่อสร้างผลลัพธ์ของงานดั้งเดิม

วิธี การforkเริ่มงานแบบอะซิงโครนัสในบางเธรด และ วิธีการ เข้าร่วมช่วยให้คุณรอให้งานนี้เสร็จสิ้น

ใหม่WorkStealingPool

วิธี การ newWorkStealingPoolมีการใช้งานสองแบบ:

public static ExecutorService newWorkStealingPool(int parallelism) {
        return new ForkJoinPool
            (parallelism,
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }

public static ExecutorService newWorkStealingPool() {
        return new ForkJoinPool
            (Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }

ตั้งแต่เริ่มต้น เราทราบว่าภายใต้ประทุน เราไม่ได้เรียกตัวสร้างThreadPoolExecutor ที่นี่เรากำลังทำงานกับเอนทิตีForkJoinPool เช่นเดียวกับThreadPoolExecutorเป็นการใช้งานAbstractExecutorService

เรามี 2 วิธีให้เลือก ในตอนแรกเราเองระบุว่าเราต้องการเห็นความเท่าเทียมในระดับใด หากเราไม่ระบุค่านี้ ความขนานของพูลของเราจะเท่ากับจำนวนแกนประมวลผลที่มีในเครื่องเสมือน Java

มันยังคงต้องเข้าใจว่ามันทำงานอย่างไรในทางปฏิบัติ:

Collection<Callable<Void>> tasks = new ArrayList<>();
        ExecutorService executorService = Executors.newWorkStealingPool(10);

        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            Callable<Void> callable = () -> {
                System.out.println("Processed user request #" + taskNumber + " on thread " + Thread.currentThread().getName());
                return null;
            };
            tasks.add(callable);
        }
        executorService.invokeAll(tasks);

เราสร้างงาน 10 รายการที่แสดงสถานะการเสร็จสิ้นของตนเอง หลังจากนั้น เราเริ่มงานทั้งหมดโดยใช้เมธอดinvokeAll

ผลลัพธ์เมื่อดำเนินการ 10 งานใน 10 เธรดในพูล:

เราเห็นว่าหลังจากสร้างคิวแล้ว เธรดจะใช้งานเพื่อดำเนินการ คุณยังสามารถตรวจสอบได้ว่าจะกระจายงาน 20 งานในกลุ่ม 10 เธรดอย่างไร

จากผลลัพธ์ เราจะเห็นว่าบางเธรดจัดการเพื่อทำงานหลายอย่างให้เสร็จ ( ForkJoinPool-1-worker-6เสร็จ 4 งาน) ในขณะที่บางเธรดทำเพียงงานเดียว ( ForkJoinPool-1-worker-2 ) หากมีการเพิ่มการหน่วงเวลา 1 วินาทีในการดำเนินการตาม วิธี การโทรรูปภาพจะเปลี่ยนไป

Callable<Void> callable = () -> {
   System.out.println("Processed user request #" + taskNumber + " on thread " + Thread.currentThread().getName());
   TimeUnit.SECONDS.sleep(1);
   return null;
};

เพื่อประโยชน์ในการทดลอง ลองเรียกใช้รหัสเดียวกันบนเครื่องอื่น ผลลัพธ์ที่ได้:

ในผลลัพธ์นี้ เป็นที่น่าสังเกตว่าเรา "ขอ" เธรดในพูล ยิ่งไปกว่านั้น ชื่อของเธรดผู้ปฏิบัติงานไม่ได้เปลี่ยนจากหนึ่งถึงสิบ แต่บางครั้งก็สูงกว่าสิบแทน เมื่อดูชื่อเฉพาะ เราจะเห็นว่ามีคนงานสิบคนจริงๆ (3, 5, 9, 13, 17, 19, 21, 23, 27 และ 31) ที่นี่ค่อนข้างสมเหตุสมผลที่จะถามว่าทำไมสิ่งนี้ถึงเกิดขึ้น? เมื่อใดก็ตามที่คุณไม่เข้าใจสิ่งที่เกิดขึ้น ให้ใช้ดีบักเกอร์

final ForkJoinPool forkJoinPool = (ForkJoinPool) executorService;

เราจะใช้การดำเนินการประเมินนิพจน์เพื่อตรวจสอบวัตถุนี้หลังจากเรียกใช้เมธอดinvokeAll ในการทำเช่นนี้ หลังจาก เมธอด invokeAllให้เพิ่มคำสั่งใดๆ เช่น sout ที่ว่างเปล่า และตั้งค่าเบรกพอยต์

เราจะเห็นว่าพูลมี 10 เธรด แต่ขนาดของอาร์เรย์ของเธรดผู้ปฏิบัติงานคือ 32 แปลก แต่ก็โอเค มาขุดกันต่อ เมื่อสร้างพูล ให้ลองตั้งค่าระดับความขนานเป็นมากกว่า 32 เช่น 40

ExecutorService executorService = Executors.newWorkStealingPool(40);

ในดีบักเกอร์ ลองดูที่วัตถุ forkJoinPool อีกครั้ง

ตอนนี้ขนาดของอาร์เรย์ของเธรดผู้ปฏิบัติงานคือ 128 เราถือว่านี่เป็นหนึ่งในการปรับให้เหมาะสมภายในของ JVM ลองค้นหาในรหัสของ JDK (openjdk-14):

อย่างที่เราสงสัย: ขนาดของอาร์เรย์ของเธรดผู้ปฏิบัติงานคำนวณโดยดำเนินการจัดการระดับบิตบนค่าความขนาน เราไม่จำเป็นต้องพยายามหาว่าเกิดอะไรขึ้นที่นี่ แค่รู้ว่ามีการเพิ่มประสิทธิภาพดังกล่าวก็เพียงพอแล้ว

อีกแง่มุมที่น่าสนใจในตัวอย่างของเราคือการใช้เมธอดinvokeAll เป็นที่น่าสังเกตว่า เมธอด invokeAllสามารถส่งคืนผลลัพธ์หรือมากกว่ารายการผลลัพธ์ (ในกรณีของเรา a List<Future<Void>>)ซึ่งเราสามารถค้นหาผลลัพธ์ของแต่ละงานได้

var results = executorService.invokeAll(tasks);
        for (Future<Void> result : results) {
            // Process the task's result
        }

บริการชนิดพิเศษและเธรดพูลสามารถใช้ในงานที่มีระดับของการทำงานพร้อมกันที่คาดเดาได้หรืออย่างน้อยก็โดยนัย