Блог

Я новичок в spring webflux, и мое текущее приложение spring boot использует планировщик (с пометкой @Scheduled) для чтения списка данных из БД, одновременного пакетного вызова rest api, а затем записи в поток событий. Я хочу добиться того же в Spring webflux.

1. Должен ли я использовать @Scheduled или использовать schedulePeriodically из Webflux?
2. Как я могу разбивать элементы из БД на меньшие наборы (скажем, 10 элементов) и одновременно вызывать rest api?
3. В настоящее время приложение извлекает максимум 100 записей за один запуск планировщика, а затем обрабатывает их. Я планирую перейти на r2dbc, если я это сделаю, могу ли я ограничить поток данных, например, 100?

Спасибо

1. Должен ли я использовать @Scheduled или использовать schedulePeriodically из Webflux?

@Scheduled это аннотация, которая является частью запланированного пакета spring framework, в то время как schedulePeriodically — это функция, которая является частью reactor, поэтому вы не можете сравнивать их. Я не вижу никаких проблем в использовании аннотации, поскольку она является частью основного фреймворка.

2. Как я могу разбивать элементы из БД на меньшие наборы (скажем, 10 элементов) и одновременно вызывать rest api?

Используя Flux#buffer функции, которые будут выдавать список элементов, когда буфер будет заполнен.

 Flux.just("1", "2", "3", "4")
        .buffer(2)
        .doOnNext(list -> {
            System.out.println(list.size());
        }).subscribe()
 

Будет печатать 2 каждый раз.

3. В настоящее время приложение извлекает не более 100 записей за один запуск планировщика, а затем обрабатывает их. Я планирую перейти на r2dbc, если я это сделаю, могу ли я ограничить поток данных, например, 100?

Ну, вы можете, как было написано ранее, извлекать, а затем буферизировать ответы в списки по 100, затем вы можете поместить каждый список в свой собственный поток и снова выдавать элементы или обрабатывать каждый список из 100 элементов. Зависит от вас.

В сегменте буфера есть много функций, ознакомьтесь с ними.

Поток #буфер

Flux.buffer объединит потоки и выдаст список потоков указанного размера буфера. Для пакетной обработки вы можете использовать Flux.expand или Mono.expand. Вам нужно только указать свое условие в расширении, чтобы выполнить его снова или окончательно завершить. Вот несколько примеров:

     public static void main(String[] args) {
        List<String> list = new ArrayList<>();
        list.add("1");
        
        
        Flux.just(list)
        .buffer(2)
        .doOnNext(ls -> {
            System.out.println(ls.getClass());
            // Buffering a list returns the list of list of String
            System.out.println(ls);
        }).subscribe();
        
        Flux.just(list)
        .expand(listObj -> {
            // Condition to finally end the batch 
            if (listObj.size()>4) {
                return Flux.empty();
            }
            // Can return the size of data as much as you require
            list.add("a");
            return Flux.just(listObj);
        }).map(ls -> {
            // Here it returns list of String which was the original object type not list of list as in case of buffer
            System.out.println(ls.getClass());
            System.out.println(ls);
            return ls;
        }).subscribe();
    }


Output:

class java.util.ArrayList
[[1]]            /// Output of buffer list of list 
class java.util.ArrayList
[1]
class java.util.ArrayList
[1, a]
class java.util.ArrayList
[1, a, a]
class java.util.ArrayList
[1, a, a, a]
class java.util.ArrayList
[1, a, a, a, a]