Comments 54
Если вы имеете в виду callbacks, которые устанавливает пользовательское приложение, то оно не имеет доступа к потоку клиентской библиотеки. Поэтому добавляться callbacks могут из неизвестных заранее потоков.
Например: https://github.com/atomix/copycat/issues/75
Дело в том, что это общепринятая практика, что клиентская либа поддерживает какой-то внутренний пул потоков, в которых производит сетевые операции. Например java-клиент Kafka так работает.
Конечно, если внимательно читать документацию, то многих проблем бы удалось избежать.
Используйте ListenableFuture из guava и забудьте об этих проблемах. Там можно явно указать executor, в котором должен выполняться listener. Да и в целом у ListenableFuture лучше продуман публичный API.
С какой стати забудьте? Указание executor вовсе не избавляет от потенциальных проблем. Оно просто гарантирует (а точнее позволяет гарантировать), в каком потоке вызовется listener. Кстати его и тут можно указать, о чем в посте было написано — и в комментах уже кажется много раз.
Налажать при этом все равно никто не помешает — и это явно написано в javadoc у ListenableFuture в одном из первых абзацев.
Я ровно об этом и пытаюсь сказать. Вот цитата автора:
"
Асинхронность в клиенте реализовывалась с использованием CompletableFuture, все операции внутри клиента производились в одном потоке (далее в коде — singleThreadExecutor)."
Автор гарантирует отсутствие гонок в неком потоконебезопасном коде выполняя его в single thread executor, но получает гонки, обращайтесь к этому самому небезопасному коду из колбэков фучи. Для решения этой проблемы существует стандартный механизм в ListenableFuture (и, как тут правильно заметили в CallableFuture и в Scala) — при добавлении колбэка указать executor, в котором он будет выполнен. Т.е. для решения проблемы автора ему было достаточно при добавлении колбэка указать тот самый single thread executor, в котором он выполняет свой небезопасный код.
Интересно, что в Future'ах скалы это проблема решается тем, что все методы, которые принимают коллбэки, также принимают параметром executor, в котором коллбэк должен выполниться, поэтому всегда можно указать нужный executor и сериализовать обращения к мутабельному состоянию. Хотя для этого лучше конечно использовать библиотеки вроде Akka.
- либо синхронизацией доступа к pendingFutures и другим внутренним объектам библиоткеи,
- либо переносом строчки pendingFutures.remove(future); из колбэка сразу за строчку future.complete(data);? Тогда добавление и удаление точно происходили бы в singleThreadExecutor.
А чем вам этот вариант CompletableFuture неполноценный?
Да ладно, где тут лишняя?
Сразу несколько задач в разных потоках — никаких проблем. ExecutorService это ровно для этого. Только надо выбрать многопоточный. Я только вот что хочу сказать — async/await в C# — это синтаксический сахар, на уровне языка, такого вы конечно в Java самостоятельно не сделаете. А во всем остальном то что есть сейчас — вполне адекватно задачам. Местами может быть есть оверинжиниринг, не без этого, но все совершенно полноценное.
Новшество конечно… в 8-ке. Только там помимо этого были еще и лямбды и стримы, коотрые сами по себе такое большое новшество, что релиз откладывали несколько раз. Я думаю просто не успели.
Можно кстати хоть на код этого CF посмотреть — на мой взгляд, там такой ад и ужас, в отличие от многих других классов. Наводит на мысли, что торопились слегка.
А насчет сразу хорошо… ну не знаю, я бы не хотел, чтобы синтаксис языка меняли из-за того, что можно сделать библиотекой. В большинстве случаев это получается более гибко. И сначала надо понять, какие кейсы все-таки больше нужны, и начинать делать с них. Вот лямбды, несмотря на все споры вокруг них, активно пошли в народ, и просто невооруженным взглядом видно, насколько стал меняться стиль во многих фреймворках, в сторону упрощения причем.
Вот тут лишняя:
pendingFutures.add(future);
future.whenComplete((v, e) -> {
pendingFutures.remove(future);
});Сравните:
pendingFutures.add(future);
await future;
pendingFutures.remove(future);Это не накладные расходы нифига. Это последствия того, что у CF много разных методов. Для разных же целей. Сделать обертку, которая тупо ждет, и ничего не делает — раз плюнуть. И будет одинаково.
Эта статься — очень сильное упрощение. Ну вот например:
Всё дело в том, что не нужно писать .then, создавать анонимную функцию для обработки ответа, или включать в код переменную с именем data, которая нам, по сути, не нужна.
Это чепуха. Не полная, но в значительной степени. Что значит нам не нужна переменная data? Во-первых, промисы можно сцеплять в цепочки, и обрабатывать ответ много раз (например, для разных потребителей). При этом каждый обработчик будет получать результат работы предыдущего. Тот сценарий, когда можно без этого — это лишь один частный случай. И пусть он даже часто встречается — но есть и другие.
И потом, это все написано для javascript. А там все сильно иначе, чем в java, как в node, так и в вебе.
А кто говорил про накладные расходы-то? Речь шла про лишнюю вложенность.
Я именно про эти накладные расходы — в виде лишних строчек. Их нет, потому что ваш пример с await — не их той оперы. Напомню:
CF — это A Future that may be explicitly completed (setting its value and status), and may be used as a CompletionStage, supporting dependent functions and actions that trigger upon its completion.
Если вам не нужен обработчик завершения (dependent functions and actions), который и есть вложенный вызов — то зачем вас сдался CompletableFuture? Возьмите обычный, и сделайте get() — и будет вам щастье.
Кажется, вы не поняли. Приведенные мною два куска псевдокода делают одно и то же. В смысле, могли бы делать если бы async/await были включены в язык.
Это вы меня не поняли. Если вам не нужен callback, который будет асинхронно вызван по завершению — то зачем вы берете CompletableFuture? Возмите просто Future, вызовите .get(), и будет у вас ровно тоже число строк — одна.
А если callback нужен — то пример с await неполный, и его надо переписать, и тогда он может будет проще и короче, а может и не будет.
Вообще вся вот эта вот фигня со списками и remove — она ровно для того, чтобы отслеживать, какие процессы завершились. А если у вас асинхронные callbacks — то вы и так знаете, когда и что завершилось, и эти же вещи удобнее делать не так. Изначальный код — он вообще странен, он от обычный Futures, так не пишут.
pendingFutures.add(future);
await future;
pendingFutures.remove(future); <- Это и есть call back. Все что после await асинхронно вызовется после завершения
Хотя пример и правда странный
Сколько у future будет лишней вложенности?
await pendingFutures.addAsync(future);
var result = await future;
var foo = result.SomeField ?? await resut.DoSecondRequestAsync()
await pendingFutures.removeAsync(future);
foo.StartMainCircle().FireAndFoget()
try {
await pendingFutures.addAsync(future);
var result = await future;
var foo = result.SomeField ?? await resut.DoSecondRequestAsync()
await pendingFutures.removeAsync(future);
foo.StartMainCircle().FireAndFoget()
} catch(Exeption ex){
//work with ex
throw ex
}
Я вам именно про это и толкую. Если у вас такая логика — вам прямая дорога к использованию просто Future, и просто .get(). У Future вообще нет callbacks, если они вам не нужны — зачем вы берете CompletableFuture, чье основное назначение — именно наличие callbacks?
Чтобы не блокировать поток!
Вы вообще пост-то читали — или сразу в комментарии полезли?
Мною разрабатывался асинхронный, неблокирующийся однопоточный клиент к серверу
Нельзя в таком сервере использовать метод .get(), вот никак нельзя!
Оператор await применяется к задаче в асинхронном методе для приостановки выполнения метода до завершения выполнения ожидаемой задачи. Задача представляет выполняющуюся работу.
Это я цитирую MSDN, если что.
По-вашему, приостановка потока — это не равно "блокирование"?
Приостановка выполнения метода — это не то же самое что приостановка выполнения потока!
В java — одно и тоже.
Вы спросили: "А чем вам этот вариант CompletableFuture неполноценный?"
Вам ответили: тем, что в текущей реализации используется на 1 уровень вложенности больше, чем было бы, если бы в Java был оператор await с поведением аналогичным тому как он себя ведет в C#.
Вы понимаете что такое сослагательное наклонение?
Блин. При чем тут пост? Я лично давно уже говорю совершенно о другом. await приостанавливает метод до завершения — тоже самое делает .get(). Хотите callback — берите CF, не хотите — get() это ровно тоже самое. Если вы не вызываете .get() — ваш текущий метод и поток (в java это одно и тоже) продолжает выполняться параллельно с тем потоком, в котором работает Future.
То что в C# из кода async метода генерируется де-факто callback — это совершенно другая история.
Вообще-то вызов .get() заблокирует поток, его нельзя использовать вместо (гипотетического) оператора await.
Хм. Вот посмотрите внимательно на свой код с await, и расскажите мне, что делает ваш поток, если асинхронная операция на момент await еще не завершилась? Какой код он выполняет?
Он выполняет, внезапно, вызов future.whenComplete. Говорю же, оба приведенных мною кода — полностью эквивалентны.
А вот когда операция завершиться, то если нет контекста треда, то возмется любой тред из пула, а если это например ui тред, то в него запущится операция в очереедь на выполнение
Для этого для начала нужен способ определить Executor, привязанный к текущему потоку — нужно какое-нибудь свойство, привязанное к потоку.
Причем модифицировать текущие реализации этого интерфейса чтобы они назначали себя созданным ими же потокам — нельзя, потому что это нарушит инкапсуляцию существующих библиотек, которые создают себе внутри приватный Executor...
Не совсем понятно зачем городить все это городище, когда есть thenRunAsync. Представьте после вас придет человек который не имеeт таких тонких познаний, и просто в ходе рефакторинга удалит двойной CompletableFuture.
Это просто ужасный костыль и зачем вообще лезть в дебри Executor-ов когда речь идет о простой асинхроной библиотеке?
В java асинхронизм построен на потоках, но это означает вы должны их пихать во все уголки вашей библиотеки. Если CompletableFuture вам не достаточно, используйте RxJava.
Проблема использования CompletableFuture в нескольких потоках и её решение