Эта концепция блокировки потока Java плоха?

Question

Я попытаюсь кратко объяснить концепцию блокировки потоков, которую я придумал на примере.Рассмотрим следующий пример программы.

public class Main {
    public static void main(String[] args) {
        Data data = new Data();

        while (true) {
            doStuff();
            doStuff();

            for (int i = 0; i < 256; i++) {
                System.out.println("Data " + i + ": " + data.get(i));
            }

            doStuff();
            doStuff();

            for (int i = 0; i < 256; i++) {
                data.set(i, (byte) (data.get(i) + 1));
            }

            doStuff();
            doStuff();
        }
    }

    public static void doStuff() {
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class Data {
    private final byte[] data = new byte[256];

    public byte get(int i) {
        return data[i];
    }

    public void set(int i, byte data) {
        this.data[i] = data;
    }
}

Важно, чтобы только основной поток изменял data.Теперь я хочу сделать цикл, который печатает data асинхронно.

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        Data data = new Data();

        while (true) {
            doStuff();
            doStuff();

            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < 256; i++) {
                        System.out.println("Data " + i + ": " + data.get(i));
                    }
                }
            });

            doStuff();
            doStuff();

            for (int i = 0; i < 256; i++) {
                data.set(i, (byte) (data.get(i) + 1));
            }

            doStuff();
            doStuff();
        }
    }

После отправки задачи в executorService основной поток теперь может работать так, как нужно.Проблема в том, что основной поток может потенциально достичь точки, в которой он изменяет data до того, как он был напечатан, но состояние data должно быть напечатано при его отправке.

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

Это решение, которое я придумал для этой проблемы.

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        Data data = new Data();
        Lock lock = new Lock(); // <---------------

        while (true) {
            doStuff();
            doStuff();

            lock.lock(); // <---------------
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < 256; i++) {
                        System.out.println("Data " + i + ": " + data.get(i));
                    }

                    lock.unlock(); // <---------------
                }
            });

            doStuff();
            doStuff();

            lock.waitUntilUnlock(); // <---------------
            for (int i = 0; i < 256; i++) {
                data.set(i, (byte) (data.get(i) + 1));
            }

            doStuff();
            doStuff();
        }
    }

public class Lock {
    private final AtomicInteger lockCount = new AtomicInteger();

    public void lock() {
        lockCount.incrementAndGet();
    }

    public synchronized void unlock() {
        lockCount.decrementAndGet();
        notifyAll();
    }

    public synchronized void waitUntilUnlock() {
        while (lockCount.get() > 0) {
            try {
                wait();
            } catch (InterruptedException e) {

            }
        }
    }
}

ТеперьОсновной поток может перейти к работе над другими вещами после отправки data.По крайней мере, может, пока не достигнет точки, где он модифицирует data.

Вопрос: это хороший или плохой дизайн?Или есть лучшая (уже существующая) реализация для этой проблемы?

Обратите внимание, что ReentrantLock не будет работать в этом случае.Я должен заблокировать перед отправкой в ​​основной поток и снять блокировку в потоке исполнителя.

Answer 1

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

В этом случае вы можете просто использовать общую очередь блокировки (синхронная очередь 1004 * мне подходит) между двумя потоками:

    ExecutorService executorService = Executors.newSingleThreadExecutor();
    Data data = new Data();
    SynchronousQueue queue = new SynchronousQueue();

    while (true) {
        doStuff();
        doStuff();

        executorService.submit(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 256; i++) {
                    System.out.println("Data " + i + ": " + data.get(i));
                }
                queue.put(data);
            }
        });

        doStuff();
        doStuff();

        data = queue.take();
        for (int i = 0; i < 256; i++) {
            data.set(i, (byte) (data.get(i) + 1));
        }

        doStuff();
        doStuff();
    }

Answer 2

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

Перепишите асинхронную часть как вызываемую, чтобы она возвращала данные в результате.

Callable<Integer> task = () -> {
    Data data = new Data();
    for (int i = 0; i < 256; i++) {
        System.out.println("Data " + i + ": " + data.get(i));
    }
    return data;
};

ExecutorService executor = Executors.newFixedThreadPool(1);
Future<Data> future = executor.submit(task);
doStuff();
// ... main thread goes about its business

// when you get to a point where you need data, 
// you can block here until the computation is done
Data data = future.get(); 

Таким образом, Future обеспечивает отображение объекта Data в потоках.

Answer 3

Я нашел альтернативный подход к проблеме, который также отвечает на мой комментарий из ответа @JB Nizet.ExecutorService#submit(Runnable) возвращает Future<?>, который можно использовать для ожидания готовности задачи.Если несколько представлений возможны, можно просто создать Queue<Future<?>> queue, всегда предлагать Future<?>, возвращаемый ExecutorService#submit(Runnable) queue, и в тот момент, когда основной поток должен ждать всего #poll().get() всего queue.

Редактировать: Я также нашел соответствующий ответ здесь: https://stackoverflow.com/a/20496115/3882565

Answer 4

Кажется, что это достаточно хорошо покрыто базовой синхронизацией: doStuff ожидает, что будет иметь единственный доступ к данным, предположительно, чтобы он мог безопасно изменять данные в целом.Между тем, doPrint рассчитывает на работу со стабильными данными.В этих двух случаях данные являются единицей состояния, к которой осуществляется доступ, и соответствующий метод блокировки заключается в синхронизации для экземпляра данных, к которому осуществляется доступ.

public void runningInThread1() {
    Data someData = getData(); // Obtain the data which this thread is using
    doStuff(someData); // Update the data
    // Looping and such omitted.
}

public void runningInThread2() {
    Data someData = getData();
    doPrint(someData); // Display the data
}

public void doStuff(Data data) {
    synchronized ( data ) {
        // Do some stuff to the data
    }
}

public void doPrint(Data data) {
    synchronized ( data ) {
        // Display the data
    }
}

В качестве альтернативы, если doStuffи doPrint реализованы как методы экземпляра Data, тогда можно было бы выполнить синхронизацию, добавив ключевое слово synchronized к методам экземпляра.