Gọi tuần tự trên luồng song song làm cho tất cả các hoạt động trước đó theo thứ tự

Question

Tôi có một bộ dữ liệu quan trọng và muốn gọi phương thức chậm, nhưng sạch sẽ và gọi phương thức nhanh với các hiệu ứng phụ trên kết quả của thao tác đầu tiên. Tôi không quan tâm đến kết quả trung gian, vì vậy tôi không muốn thu thập chúng.

Giải pháp rõ ràng là tạo luồng song song, thực hiện cuộc gọi chậm, tạo luồng liên tiếp một lần nữa và thực hiện cuộc gọi nhanh. Vấn đề là, TẤT CẢ mã thực hiện trong chủ đề duy nhất, không có song song thực tế.

Ví dụ mã:

@Test 
public void testParallelStream() throws ExecutionException, InterruptedException 
{ 
    ForkJoinPool forkJoinPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors() * 2); 
    Set<String> threads = forkJoinPool.submit(()-> new Random().ints(100).boxed() 
      .parallel() 
      .map(this::slowOperation) 
      .sequential() 
      .map(Function.identity())//some fast operation, but must be in single thread 
      .collect(Collectors.toSet()) 
    ).get(); 
    System.out.println(threads); 
    Assert.assertEquals(Runtime.getRuntime().availableProcessors() * 2, threads.size()); 
} 

private String slowOperation(int value) 
{ 
    try 
    { 
     Thread.sleep(100); 
    } 
    catch (InterruptedException e) 
    { 
     e.printStackTrace(); 
    } 
    return Thread.currentThread().getName(); 
} 

Nếu tôi loại bỏ sequential, mã thực hiện như mong đợi, nhưng, rõ ràng, hoạt động không song song sẽ được gọi theo nhiều chủ đề.

Bạn có thể giới thiệu một số tham chiếu về hành vi đó hay có thể một số cách để tránh các bộ sưu tập tạm thời?

Answer 1

Chuyển luồng từ parallel() thành sequential() hoạt động trong thiết kế API dòng ban đầu, nhưng gây ra nhiều sự cố và cuối cùng là triển khai là changed, do đó, nó chỉ bật và tắt cờ song song cho toàn bộ đường ống. Các tài liệu hiện nay là thực sự mơ hồ, nhưng nó đã được cải thiện trong Java-9:

các đường ống dẫn dòng được thực hiện tuần tự hoặc song song tùy thuộc vào chế độ của dòng mà trên đó các hoạt động thiết bị đầu cuối được gọi. Chế độ tuần tự hoặc song song của luồng có thể được xác định bằng phương pháp BaseStream.isParallel() và chế độ của luồng có thể được sửa đổi với các hoạt động BaseStream.sequential() và BaseStream.parallel(). Cài đặt chế độ tuần tự hoặc song song gần đây nhất áp dụng cho việc thực thi toàn bộ luồng luồng.

Đối với vấn đề của bạn, bạn có thể thu thập tất cả mọi thứ vào trung List và bắt đầu đường ống tuần tự mới:

new Random().ints(100).boxed() 
     .parallel() 
     .map(this::slowOperation) 
     .collect(Collectors.toList()) 
     // Start new stream here 
     .stream() 
     .map(Function.identity())//some fast operation, but must be in single thread 
     .collect(Collectors.toSet()); 

Answer 2

Trong triển khai hiện tại, Luồng hoặc là tất cả song song hoặc tất cả tuần tự. Trong khi Javadoc không rõ ràng về điều này và nó có thể thay đổi trong tương lai, điều này có thể nói điều này là có thể.

S song song()

Trả về một dòng tương đương đó là song song. Có thể trả lại chính nó, hoặc bởi vì dòng đã được song song, hoặc vì trạng thái dòng cơ bản đã được sửa đổi để được song song.

Nếu bạn cần chức năng là chuỗi đơn, tôi khuyên bạn nên sử dụng khối hoặc phương pháp khóa hoặc đồng bộ hóa.