Как конвертировать события Java Swing в Kotlin Flow с максимальной производительностью
Я пытаюсь преобразовать события Java Swing, обычно получаемые с помощью прослушивателей событий, в поток этих событий Kotlin. Я проверил код многих проектов и нашел несколько способов сделать это. Мне было интересно, может ли кто-нибудь дать мне причины использовать один стиль вместо другого в отношении производительности/надежности/использования ресурсов:
fun JButton.actionEvents1(): Flow<ActionEvent> = callbackFlow {
val listener = ActionListener { e ->
trySend(e)
}
addActionListener(listener)
awaitClose {
removeActionListener(listener)
}
}
fun JButton.actionEvents2(): Flow<ActionEvent> = callbackFlow {
val listener = ActionListener { e ->
trySend(e).isSuccess
}
addActionListener(listener)
awaitClose {
removeActionListener(listener)
}
}
fun JButton.actionEvents3(): Flow<ActionEvent> = callbackFlow {
val listener = ActionListener { e ->
trySendBlocking(e)
}
addActionListener(listener)
awaitClose {
removeActionListener(listener)
}
}
fun JButton.actionEvents4(
scope: CoroutineScope
): Flow<ActionEvent> = callbackFlow {
val listener = ActionListener { e ->
scope.launch { send(e) }
}
addActionListener(listener)
awaitClose {
removeActionListener(listener)
}
}
В последнем примере предполагается, что у нас есть доступ к экземпляру CoroutineScope, который в моем случае будет той же областью, из которой получается поток событий каждого действия и вызывается collect. Следует ли мне избегать запуска сопрограммы для каждого события?
Я тоже надеялся использовать ActionListener { e -> send(e) }, но в Котлине это запрещено Suspension functions can be called only within coroutine body.
Ответы 2
Какой из них использовать, зависит от того, как вы хотите поступать в ситуациях, когда ActionEvent выдаются быстрее, чем их можно обработать.
callbackFlow использует внутренний канал для разделения отправки и получения событий, действуя как блокирующая очередь. События из ActionListener отправляются в канал и хранятся там до тех пор, пока не будут получены сборщиком потока. Когда поток собирается медленнее, чем создаются новые события, в канале создается противодавление, которое поглощается буфером канала. Буфер, однако, ограничен (по умолчанию до 64 элементов), и когда буфер заполнен, новые события больше не могут отправляться.
Вам необходимо указать, что должно произойти в данной ситуации. Есть несколько вариантов:
- Блокируйте текущий поток, пока не освободится место.
- Приостановить текущий поток, пока не освободится место.
- Пропустить текущее событие.
- Удалите самое старое событие из буфера, чтобы освободить место для нового события.
Прежде чем вдаваться в подробности, вы всегда можете увеличить размер буфера, предотвращая возникновение проблемных ситуаций с противодавлением. Используйте .buffer(100) (или любой другой размер буфера, который вам кажется подходящим) на callbackFlow { ... }. Однако будьте осторожны: если вы постоянно генерируете события быстрее, чем они обрабатываются, это только отсрочит возникновение проблемы. Увеличивайте размер буфера только в том случае, если вы можете успешно поглощать всплески, когда множество событий генерируется пакетами.
Вариант 1 (Блокировать): trySendBlocking вашей actionEvents3 функции блокирует основной поток, который использовался для выдачи события, до тех пор, пока канал больше не переполнится. Пока основной поток заблокирован, ваш пользовательский интерфейс не отвечает. Вам следует использовать это только в том случае, если на канале не будет большого обратного давления (т. е. события обычно обрабатываются достаточно быстро до того, как будет создано следующее событие).
Вариант 2 (Приостановить): send вашей функции actionEvents4 приостанавливает текущий поток до тех пор, пока канал не перестанет заполняться. Приостановка основного потока была бы полезна, поскольку она позволяет другому коду выполняться в основном потоке, сохраняя отзывчивость вашего пользовательского интерфейса. Однако здесь может быть приостановлен даже не основной поток, поскольку единственное, что вы делаете в этом потоке, — это запускаете новую сопрограмму. Эта сопрограмма приостанавливается send, и эта сопрограмма запускается в предоставленном вами scope. Какой из них может использовать или не использовать основной поток, не имеет значения.
Этот вариант самый безопасный, но и самый сложный. Если у вас серьезные проблемы с обратным давлением и вы не можете пропустить или удалить какие-либо события, это все равно может быть самым чистым решением.
Вариант 3 (Пропустить): trySend вашей actionEvents1 функции отказывается от отправки события, если канал заполнен, фактически пропуская новое событие. Является ли такое поведение подходящим, зависит от характера событий. Повторное нажатие одной и той же кнопки, настолько быстрое, что событие щелчка невозможно обработать до того, как произойдет следующий щелчок, может быть приемлемой ситуацией, когда самый новый щелчок можно игнорировать.
Ваша функция actionEvents2 ведет себя идентично actionEvents1. Единственная разница заключается в том, что вы получаете доступ к isSuccess, определяя, было ли новое событие отправлено успешно. Но вы ничего не делаете с этой информацией и немедленно завершаете лямбду, поэтому не имеет значения, была ли она на самом деле успешной или нет. Не следует использовать actionEvents2.
Вариант 4 (удаление): добавьте .buffer(onBufferOverflow = BufferOverflow.DROP_OLDEST) к callbackFlow { ... } (здесь вы также можете указать размер буфера) и используйте trySend для отправки новых событий. Если канал заполнен, самое старое событие будет удалено, и можно будет немедленно вставить новое событие. Это может быть полезно, если вам не нужны старые события и вы хотите обрабатывать только самые новые события. В этом случае вам может потребоваться фактически уменьшить размер буфера, чтобы принудительно удалить все старые события вместо того, чтобы сохранять старые события в том виде, в котором они помещаются в буфер. Также посмотрите, хотите ли вы объединить поток в этой ситуации.
Я предполагаю, что в коде, который я заметил в некоторых проектах с открытым исходным кодом, trySend(e).isSuccess появляется только потому, что это эквивалент старой функции offer, которая сейчас устарела, а сообщение об устаревании гласит: replaceWith = ReplaceWith("trySend(element).isSuccess"). Действительно, она не представляет никакой ценности.
Не существует единственного правильного способа сделать это, потому что ответ зависит от того, какое поведение вы ищете. Вам нужно решить, что вы хотите, чтобы произошло в том маловероятном случае, если кнопка будет нажата повторно, быстрее, чем приложение сможет на нее отреагировать.
На самом деле есть три возможных решения.
- Отбросьте лишние события. Хотите ли вы такого поведения, будет зависеть от того, что на самом деле делает кнопка. Вы можете добиться такого поведения, объединив
trySendс небольшим или несуществующим буфером. Некоторые клики будут игнорироваться, когда приложение занято. - Заблокируйте пользовательский интерфейс, пока программа не догонит его. Инстинктивно это может показаться неправильным: блокировать поток пользовательского интерфейса — это плохо. Но на самом деле это может быть жизнеспособным решением, поскольку оно не позволит пользователю вводить дополнительные данные до тех пор, пока приложение не будет готово их обработать. Это то, что вы получите, если будете использовать
trySendBlocking, когда буфер потока заполнен. - Буферизируйте события в очереди. Вы можете получить такое поведение, если объедините
trySendилиtrySendBlockingс достаточно большим буфером. Каждый щелчок в конечном итоге будет обработан один за другим.
Размер буфера потока можно изменить, связав вызов buffer() или conflate().
Разумная стратегия, вероятно, представляет собой гибрид:
- Добавьте буфер, достаточно большой, чтобы обработать ожидаемое количество событий. Поскольку мы имеем дело с нажатиями кнопок, это, вероятно, не будет большим.
- Если буфер заполняется, либо заблокируйте пользовательский интерфейс, либо удалите события.
Как только вы решите, какое поведение вам нужно, реализация в большинстве случаев выбирает сама.
Вы заметите, что я не упомянул версию launch { send(…) }. Фактически, это всего лишь вариант использования большого/неограниченного буфера. Если новая сопрограмма не может немедленно отправить свое событие, она приостанавливается и ждет в очереди отправки, пока не сможет. Итак, буфер все еще есть; он просто скрыт в механизме сопрограммы. Использование явного buffer() в потоке делает намерение более ясным и (предположительно), в любом случае, вероятно, будет более производительным.
Обратите внимание: если вы хотите использовать launch(), вам не нужно указывать область действия, поскольку у callbackFlow уже есть своя.
Есть причины не просто делать
ActionListener { e -> send(e) }? Потому чтоtrySendотменит событие, если канал окажется полным. Выполнениеsend, по крайней мере, приведет к приостановке работы сопрограммы до тех пор, пока ее можно будет обработать. Вероятно, это могло бы привести к зависанию пользовательского интерфейса, но не лучше ли это, чем отбрасывать события при перегрузке? Я думаю, зависит от вашего варианта использования.