Как мне создать Flux / Publisher для потоковой передачи данных

Question

Я использую метод опроса для периодической выборки данных. Новые данные могут поступить в любое время. Я хочу выставить реактивный интерфейс моему клиенту. Итак, я хочу создать издателя (Flux?), Который будет публиковать новые данные, когда они станут доступны, и уведомлять об этом подписчиков. Как я могу это сделать? Все примеры Flux, которые я вижу, относятся к случаям, когда данные уже известны / доступны. По сути, я хочу что-то вроде Flux, основанного на очереди, и мой поток опроса может продолжать заполнять очередь, когда находит новые данные.

Answer 1

Для чего-то простого, вы можете использовать DirectProcessor.Это не самый сложный приемник флюса, но он поможет вам в этом.

Я написал небольшой пример, извините, он в kotlin.

val hot = DirectProcessor.create<String>()
hot.onNext("Hello")//not printed
hot.subscribe { println(it) }
//Java: hot.subscribe((it) -> System.out.println(it));


hot.onNext("Goodbye")//printed
Thread.sleep(100)
hot.onNext("foo")//printed

DirectProcessor реализует Flux, поэтому вы можете использовать его как Flux.

Как видите, элементы, добавленные до подписки на горячий источник, не будут переданы подписке.

Глядя на другие посты, Flux # create и Flux # generate также могут быть хорошими местами для начала. Разница между Flux.create и Flux.generate Это увеличит сложность и контроль над потоком.