Принудительная последовательная оценка для не поточно-ориентированного потребителя в Java

Question

С помощью API потока Java8 можно ли как-то принудительно заставить поток обрабатываться последовательно от потребителя?

Когда у меня есть java.util.function.Consumer, который, как я знаю, не является потокобезопасным, я бы хотел принудительно вызватьAPI-интерфейс Streams обрабатывает его последовательно, поскольку параллельная работа «всегда» приводит к ошибочному поведению.

Этот фрагмент иллюстрирует мою проблему:

static class NonThreadsafeConsumer<T> implements Consumer<T> {

    @Override
    public void accept(T arg0) {
        //Do non-threadsafe stuff here
    }
}

public void doIt(Stream<String> stream) {
    //Unknown behaviour
    stream.forEach(new NonThreadsafeConsumer<>());
    // Bug for sure
    stream.parallel().forEach(new NonThreadsafeConsumer<>());
    // Correct
    stream.sequential().forEach(new NonThreadsafeConsumer<>());
}

Проблема, с которой я столкнулся, заключается в том, что япоскольку автор NonThreadsafeConsumer не хочет доверять разработчику метода doIt() всегда знать и не забывать вставлять .sequential() вызов

(обратите внимание, что обеспечение безопасности потока пользователя не является цельюмой вопрос, я просто хотел бы знать, можно ли это сделать)

Answer 1

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

Но Consumer не несет ответственности за обеспечение однопоточного использования. В Java отсутствие безопасности потоков является нормой для изменяемых классов, скажем, StringBuilder, ProcessBuilder, ArrayList, HashMap, любого вида итератора, DecimalFormat, чтобы назвать некоторые примеры широко используемых изменяемых классов, которые не являются поточно-ориентированными и не обеспечивают однопоточное использование.

Обратите внимание, что вы можете просто добавить synchronized к методу accept потребителя, чтобы обеспечить выполнение одного потока за раз. При использовании в последовательном контексте есть вероятность, что оптимизатор JVM устраняет связанные с этим издержки.

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

Answer 2

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

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

Это не самая лучшая работа, но она может помочь разработчику doIt() не упустить последовательный поток.

Надеюсь, это поможет или, по крайней мере, даст вам некоторые идеи о том, как реализовать это.

public static void main(String[] args) {
    List<String> strings = Arrays.asList("a", "b", "b");

    strings.stream()
            .forEach(ConsumerFactory.getConsumer(false));
}

static class ConsumerFactory {
    static Consumer getConsumer(boolean isPar) {
        if (isPar) {
            //Handle Parallel
        }

        return new NonThreadsafeConsumer<>();
    }
}

static class NonThreadsafeConsumer<T> implements Consumer<T> {
    @Override
    public void accept(T t) {
        //Do non-threadsafe stuff here
    }
}