Неочевидные боли многопоточности Java

21:12:41, 10 Nov 2019

Все говорят, что в Java превосходная поддержка многопоточности, я хочу показать что есть некоторое количество боли, и неочевидных решений в этой поддержке.

Тут я привожу не все примеры боли, например про безопасную публикацию объектов, безопасное конструирование [4], про happens-before, про spurious wakeup(wait() - notify()), про некорректность инкремента оператором ++ поля, в том числе volatile, про 2 потокобезопасных ленивых синглтона, ради одного из которых пришлось[15] пофиксить[16] Java Memory Model

Поток нельзя прервать

Это все знают, но пусть будет на всякий случай:

import java.util.Date;

import java.util.concurrent.TimeUnit;

public class ThreadPainNoInterrupt {

public static void main(String[] args) throws InterruptedException {

Thread t = new Thread(() -> {

while(true) {

System.out.println("" + new Date() + " ololo am a bad boy");

}

});

t.start();

TimeUnit.SECONDS.sleep(2);

t.interrupt();

}

...этот код никогда сам не остановится.

Решение - использовать встроенный флаг interrupted

import java.util.Date;

import java.util.concurrent.TimeUnit;

public class ThreadPainNoInterruptFix {

public static void main(String[] args) throws InterruptedException {

Thread t = new Thread(() -> {

while(!Thread.currentThread().isInterrupted()) {

System.out.println("" + new Date() + " ololo am a bad boy");

}

});

t.start();

TimeUnit.SECONDS.sleep(2);

t.interrupt();

}

...с которым есть некоторые нюансы - а именно, методы (ниже метод sleep()), бросающие

InterruptedException

С ним(и) есть частое заблуждение:

import java.util.Date;

import java.util.concurrent.TimeUnit;

public class ThreadPainNoInterruptSleep {

public static void main(String[] args) throws InterruptedException {

Thread t = new Thread(() -> {

while(!Thread.currentThread().isInterrupted()) {

System.out.println("" + new Date() + " ololo am a bad boy");

try {

TimeUnit.SECONDS.sleep(1);

} catch (InterruptedException e) {

e.printStackTrace();

}

});

t.start();

TimeUnit.SECONDS.sleep(2);

t.interrupt();

}

... это код, который нам помогает написать IntelliJ IDEA. Он тоже никогда не остановится.

Дело в том, что InterruptedException обладает особым смыслом, это не просто исключение[1], которое можно "проглотить" или залогировать, а сигнал нам (разработчику) что когда поток висел (state=WAITING или TIMED_WAITING) на блокирующей операции (метод sleep()) он был прерван каким-то другим потоком, и, сейчас можно/нужно сделать какие-либо завершающие операции(например удалить документы из mongodb, очистить временные файлы, и т. д.), а затем как только завершающие операции сделаны - мы действительно можем прервать поток, выставив ему флаг interrupted самостоятельно. Иными словами, когда мы поймали InterruptedException - флаг interrupted не выставлен, и мы должны сделать завершающие операции, и не забыть его выставить самостоятельно:

import java.util.Date;

import java.util.concurrent.TimeUnit;

public class ThreadPainNoInterruptSleepFix {

public static void main(String[] args) throws InterruptedException {

Thread t = new Thread(() -> {

while(!Thread.currentThread().isInterrupted()) {

System.out.println("" + new Date() + " ololo am a bad boy");

try {

TimeUnit.SECONDS.sleep(1);

} catch (InterruptedException e) {

// make clean tasks here

Thread.currentThread().interrupt();

}

});

t.start();

TimeUnit.SECONDS.sleep(2);

t.interrupt();

}

... это работает.

Интересно, что неудобство InterruptedException заметил [13] Рон Пресслер, делая Project Loom, но конкретных предложений пока нет. Looking forward to.

Почему статус interrupted сбрасывается внутри потока и мы должны выставить его обратно сами как в коде выше? Почему нельзя было сделать проще, чтобы не было неявного сброса и последующей ручной установки сего замечательного флага ?

Как объясняет Егор Бугаенко [14], могут быть завершающие действия внутри блока catch (InterruptedException e), которые вызывают например Thread.sleep() (или другие "interrupted aware" методы), который так сделан что не будет спать, если у потока interrupted=true, а быстро выбросит InterruptedException. Получается, если бы статус не сбросился, мы бы не смогли например поспать в завершающем действии. Вроде редкий кейс, но при проектировании многопоточности Java его учли.

Пример кода со StackOverflow, демонстрирущий сказанное в абзаце выше (что Thread.sleep непременно бросит исключение, если у потока стоит флаг interrupted), а также что простановка флага interrupted сократит время отмены Future

import java.util.concurrent.*;

public class ThreadStopper {

public static void main(String... ignored) throws InterruptedException, ExecutionException {

ExecutorService es = Executors.newSingleThreadExecutor();

Future<?> future = es.submit(new Runnable() {

@Override

public void run() {

methodA();

}

});

Thread.sleep(1000);

future.cancel(true); // interrupts task.

long start = System.nanoTime();

try {

future.get();

} catch (CancellationException expected) {

// ignored

}

es.shutdown();

es.awaitTermination(1, TimeUnit.MINUTES);

long time = System.nanoTime() - start;

System.out.printf("Time to cancel/shutdown was %.1f milli-seconds%n", time / 1e6);

}

private static void methodA() { // doesn't throw any checked exception

for (int i = 0; i < 100; i++)

methodB();

}

private static void methodB() {

// uncomment if you don't want to see multiple InterruptedException

// if (!Thread.currentThread().isInterrupted()) {

try {

Thread.sleep(500);

} catch (InterruptedException ie) {

System.out.println("catch InterruptedException");

// comment it and you have increase wait time up to 50 sec

Thread.currentThread().interrupt();

}

// }

}

Возникает вопрос, а можно ли, вызывая Thread.currentThread().interrupt() в одной таске "испортить" флаг у потока - так что последующие таски не выполнятся ?

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

public class CooperativeInterruptedNotFail {

public static void main(String[] args) throws InterruptedException {

final ExecutorService es = Executors.newFixedThreadPool(2);

List<Future<?>> futures = new ArrayList<>();

for (int i=0; i<10; ++i) {

futures.add(es.submit(() -> {

if (!Thread.currentThread().isInterrupted()) {

System.out.println("Make heavy computations and sleep "+Thread.currentThread().toString());

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

System.out.println("Interrupting "+Thread.currentThread().toString());

Thread.currentThread().interrupt();

}

} else {

System.out.println("Skipping task because thread "+Thread.currentThread().toString()+" is considered interrupted");

}

}));

}

TimeUnit.MILLISECONDS.sleep(500);

futures.get(0).cancel(true);

futures.get(1).cancel(true);

es.shutdown();

}

У меня не получилось, все таски завершились, несмотря на то что первые 2 были отменены и проставили флаг interrupted.

Make heavy computations and sleep Thread[pool-1-thread-2,5,main]

Make heavy computations and sleep Thread[pool-1-thread-1,5,main]

Interrupting Thread[pool-1-thread-2,5,main]

Interrupting Thread[pool-1-thread-1,5,main]