Привет, Хабр! На связи Андрей Зяблин, Java разработчик компании «Магнит». Несколько лет назад, продляя дочери подписку на Netflix, я заинтересовался, как они поддерживают обслуживание сотен миллионов клиентов. Изучив вопрос, я увлёкся реактивным подходом в программировании. Через некоторое время это увлечение принесло практическую пользу при разработке сервисов в «Магните». В статье я расскажу про три решения, которые позволяют реализовать потоковый обмен данными из БД между распределёнными приложениями:

1. Реализация с использованием hibernate

2. Реализация с использованием mybatis

3. Ограничение скорости обмена с использованием механизма обратного давления («backpressure») и библиотеки Bucket4j

Постановка задачи

Наша команда разрабатывала приложение для интеграции различных сервисов сети, а также для интеграции с государственными сервисами в области ветеринарии «ВетИС».

Упрощенно структура приложения выглядит так:

Бизнес‑запросы — запросы в формате json, предназначенные для обмена на уровне бизнес‑данных (операции, позиции операций, пр.) и раскладывающиеся в более низкоуровневые запросы в форматах, поддерживающихся госсервисами.

ЦСМ (Централизованный сервис сети «Меркурий») — разрабатываемое нами приложение интеграции.

BDSM‑CLIENT — работа с бизнес‑запросами БД СМ (базы данных сети магазинов).

MERCURY — обеспечивает взаимодействие со шлюзом к «ВетИС».

PROCESSING — выполняет обработку бизнес‑запросов до отправки запросов в шлюз.

Проблемным местом оказалась передача данных из ЦСМ в БД СМ: объёмы передаваемых данных огромны, а на стороне БД СМ напрашивался тонкий клиент, пишущий непосредственно в базу.

Требования к обмену:

• Данные извлекаются из БД

• Объём данных — миллионы записей за сеанс

• Передача ведётся через интернет по http

• Необходимо использовать ORM

Изучив различные технологии, мы пришли к мнению, что в нашем случае оптимально использовать реактивность.

Доработки на стороне ЦСМ и БД СМ:

• В синхронный код ЦСМ встраивается реактивный код обмена

• Для приёма данных на стороне БД СМ пишется отдельное реактивное приложение

Используемые технологии:

• Java streams 

• Hibernate streamResult 

• Reactive streams 

• Spring webflux(reactor) 

• Spring webclient 

• Rsocket (для демонстрации) 

Граничные условия:

• Не всегда имеется возможность использовать R2DBC. На момент разработки для Oracle не было стабильного реактивного драйвера.

• Hibernate streamResult требует активную транзакцию на всё время фетча. Нужно отслеживать завершение фетча, ошибки, закрытие соединения, чтобы завершить транзакцию. Стандартные решения в виде try‑catch не подходят в данной ситуации.

Решение с использованием hibernate

Основная сложность реализации заключалась в корректной работе с транзакциями в асинхронном режиме. Для этого пришлось создавать отдельный механизм, благо Reactor это позволяет. Основу решения составляют:

• Stateless session 

• StreamResult 

• Генерация Flux из Stream 

Далее рассмотрим код. 

Сущности

• Customer — справочник заказчиков. 

• Operation — операции с заказчиками, например, продажи. 

public class Customer { 
 
    @Id 
    @GeneratedValue(strategy = GenerationType.SEQUENCE, generator = "sequence") 
    @SequenceGenerator(name = "sequence", sequenceName = "SQ_CUSTOMER", allocationSize = 1) 
    private Long id; 
 
    @Column(name = "NAME") 
    private String name; 
 
} 
public class Operation { 
 
    @Id 
    @GeneratedValue(strategy = GenerationType.SEQUENCE, generator = "sequence") 
    @SequenceGenerator(name = "sequence", sequenceName = "SQ_OPERATION", allocationSize = 1) 
    private Long id; 
 
    @Column(name = "TOTAL_SUM") 
    private Long totalSum; 
 
    @ManyToOne 
    @JoinColumn(name = "customer_id") 
    private Customer customer; 
 
    @ManyToMany 
    @JoinTable( 
        name = "operation_detail", 
        joinColumns = @JoinColumn(name = "operation_id"), 
        inverseJoinColumns = @JoinColumn(name = "product_id")) 
    private List<Product> products; 
 
} 
 

Загрузка данных 

Загрузка данных достаточно проста и не совсем имеет отношение к теме статьи, здесь демонстрируется мощь функционального программирования. На данном примере показано, насколько функциональна библиотека reactor и как она может облегчить разработку.

Класс com.magnit.flux.common.service.InitDataService

private void fillCustomer() { 
    Flux.range(0, CUSTOMER_RECORD_LIMIT) 
        .map(i -> Customer.builder().name("Customer " + i).build()) 
        .subscribe(customerRepository::save); 
} 
 
private void fillProduct() { 
    Flux.range(0, PRODUCT_RECORD_LIMIT) 
        .map(i -> Product.builder().name("Product " + i).build()) 
        .subscribe(productRepository::save); 
} 
 
private void fillOperation() { 
    Supplier<List<Product>> productSupplier = () -> new Random().ints(5, 1, 8) 
        .boxed().distinct() 
        .map(productId -> Product.builder().id((long) productId).build()) 
        .collect(Collectors.toList()); 
    List<Customer> customers = (List<Customer>) customerRepository.findAll(); 
    Flux.range(0, OPERATION_RECORD_LIMIT) 
        .map(i -> Operation.builder() 
            .totalSum((long) random.nextInt(100)) 
            .customer(customers.get(random.nextInt(customers.size() - 1))) 
            .products(productSupplier.get()) 
            .build()) 
        .buffer(1000) 
        .subscribe(operationRepository::saveAll); 
} 

Использование hibernate для стриминга 

Класс выполняет основную работу по управлению транзакциями в асинхронном режиме. Это что‑то вроде AutoCloseable в синхронных приложениях.

Класс com.magnit.flux.hibernate.dao.HibernateFluxResultProducer

@RequiredArgsConstructor 
public class HibernateFluxResultProducer<T> { 
     
   private final EntityManagerFactory entityManagerFactory; 
 
    private StatelessSession statelessSession; 
 
    /** 
     * Функция выполняет hql запрос и создаёт Flux из Result Stream 
     */ 
    public Flux<T> execute(String qlString, Class<T> resultClass) { 
        Optional.ofNullable(statelessSession).ifPresent(e -> { 
            throw new RuntimeException("Cursor already open"); 
        }); 
        val sessionFactory = entityManagerFactory.unwrap(SessionFactory.class); 
        //Использование Stateless Session позволяет ускорить чтение за счёт использования detached объектов 
        statelessSession = sessionFactory.openStatelessSession(); 
        statelessSession.getTransaction().begin(); 
        val query = statelessSession.createQuery(qlString, resultClass); 
        //Используется возможность hibernate для потокового чтения из БД - getResultStream 
        return Flux.fromStream(query.getResultStream()); 
    } 
 
    public Mono<Void> commit() { 
        return close(true); 
    } 
 
    public Mono<Void> rollback() { 
        return close(false); 
    } 
 
    public Mono<Void> close(boolean commitTran) { 
        if (statelessSession.getTransaction().isActive()) { 
            if (commitTran) { 
                statelessSession.getTransaction().commit(); 
            } else { 
                statelessSession.getTransaction().rollback(); 
            } 
 
        } 
        statelessSession.close(); 
        return Mono.empty(); 
    } 
 
 
} 

Данный класс стартует транзакцию и выполняет hql запрос. Также он имеет методы commit и rollback.

Контроллер, работающий с hibernate 

Основной функционал работы с потоками реализован в приватном методе getOperationFlux. Генерация потока осуществляется вызовом метода Flux.usingWhen, который можно назвать асинхронным AutoCloseable. Контроллер предоставляет два эндпоинта:

• getOperationsStream — json стриминг 

• getOperationsWs — веб-сокет 

Класс com.magnit.flux.hibernate.controller.HibernateOperationController

public class HibernateOperationController { 
 
    private final ObjectFactory<HibernateFluxResultProducer<Operation>> streamResultProducerObjectFactory; 
 
    @GetMapping(path = "/operations-stream", produces = "application/stream+json") 
    public Flux<Operation> getOperationsStream() { 
        return getOperationFlux(); 
    } 
 
    @MessageMapping("operations") 
    public Flux<Operation> getOperationsWs() { 
        return getOperationFlux(); 
    } 
 
    private Flux<Operation> getOperationFlux() { 
        Mono<HibernateFluxResultProducer<Operation>> streamResultExecutorMono = Mono 
            .just(streamResultProducerObjectFactory.getObject()); 
        return Flux.usingWhen(streamResultExecutorMono, 
            se -> se.execute("select o from Operation o JOIN FETCH o.customer c", Operation.class), 
            HibernateFluxResultProducer::commit); 
    } 
 
} 

Результат работы первого метода можно посмотреть непосредственно в браузере:

Можно останавить загрузки или даже закрыть браузер. Можно убедиться, что метод close всегда вызывается.

Второй метод — более интерсный. На его основе можно реализовать обмен по протоколу Rsocket. Для тестирования Rsocket клиент реализован на javascript. Также данный пример демонстрирует использование механизма backpressure (обратное давление).

statis/index.js

client.connect().subscribe({ 
    onError: error => console.error(error), 
    onSubscribe: cancel => { 
    }, 
    onComplete: socket => { 
        socket.requestStream({ 
            data: null, 
            metadata: String.fromCharCode("operations".length) + "operations" 
        }) 
        .subscribe({ 
            onComplete: () => console.log("requestStream done"), 
            onError: error => { 
                console.log("got error with requestStream"); 
                console.error(error); 
            }, 
            onNext: value => { 
                document.getElementById( 
                    "operation").innerText = "Operation ID: " 
                    + value.data.id + ". Customer: " + value.data.customer.name; 
                //Управление обратным давлением вручную. Можно поставить на паузу (функция pause), можно возобновить чтение (функция suspend) 
                if (!paused) { 
                    subscription.request(1); 
                } 
            }, 
            onSubscribe: sub => { 
                subscription = sub; 
                subscription.request(1); 
            } 
        }); 
    } 
}); 

Результаты работы также можно посмотреть в браузере. 

Кнопки Pause и Suspend позволяют остановить и возобновить загрузку потока. 

Отношения один ко многим и многие ко многим (OneToMany &ManyToMany) 

Теперь рассмотрим достаточно частую ситуацию, когда со стороны заказчика выдвигаются новые требования. В данном случае упрощенный пример: заказчик выдвинул новые требования к обмену: теперь нужно выгружать товары, содержащиеся в операциях. Теоретически это можно сделать и с hibernate, но возникают проблемы с производительностью. Как альтернативу мы рассмотрели MyBatis. Ниже сравниваем MyBatis и hiberabte.

MyBatis как более быстрая альтернатива hibernate

В нашем случае важна способность MyBatis обрабатывать отношения один ко многим и многие ко многим в стримах. 

Стриминг данных в MyBatis 

• В MyBatis 3.4.0 добавлен тип курсор — org.apache.ibatis.cursor.Cursor, аналогичный классу Resultset в JDBC 

• Курсор реализует интерфейс итератора, поэтому использовать его очень просто 

• Курсор позволяет использовать коллекции в результатах, поэтому можно реализовать отображение отношений one-to-many и many-to-many 

Изменения в сущностях 

• Product — справочник продукции 

• Operation — добавилась связь @ManyToMany со справочником продукции 

@Entity 
@Table(name = "PRODUCT") 
@Getter 
@Setter 
@Builder 
@NoArgsConstructor 
@AllArgsConstructor 
@ToString 
public class Product { 
 
    @Id 
    @GeneratedValue(strategy = GenerationType.SEQUENCE, generator = "sequence") 
    @SequenceGenerator(name = "sequence", sequenceName = "SQ_OPERATION_DETAIL", allocationSize = 1) 
    private Long id; 
 
    @Column(name = "NAME") 
    private String name; 
 
} 
 

 

@ManyToMany 
@JoinTable( 
    name = "operation_detail", 
    joinColumns = @JoinColumn(name = "operation_id"), 
    inverseJoinColumns = @JoinColumn(name = "product_id")) 
private List<Product> products; 

Маппинг сущностей 

Основная тонкость работы с MyBatis заключается в маппинге .

<mapper namespace="com.magnit.flux.mybatis.mapper.OperationMapper"> 
 
    <select id="getAllOperations" resultMap="operationResult" resultOrdered="true"> 
        select d.operation_id id, d.product_id, p.name product_name from operation_detail d inner 
        join product p on p.id = d.product_id order by d.operation_id 
    </select> 
 
 
    <resultMap id="operationResult" type="com.magnit.flux.model.entity.Operation"> 
        <id property="id" column="id"/> 
        <collection property="products" ofType="com.magnit.flux.model.entity.Product"> 
            <id property="id" column="product_id"/> 
            <result property="name" column="product_name"/> 
        </collection> 
    </resultMap> 
 
</mapper>

Здесь getAllOperations определяет отсортированный набор данных (resultOrdered=”true”), а MyBatis автоматом группирует данные по operation.id и возвращает операции с вложенными коллекциями продукции. 

Использование MyBatis для стриминга 

Класс аналогичный HibernateFluxResultProducer. Он имеет пару отличий: 

• В качестве параметра передаётся не hql запрос, а лямбда-выражение, возвращающее курсор 

• В данном подходе не используются долгоиграющие транзакции, что даёт небольшое преимущество перед hibernate. По крайней мере DBA будут спокойнее).

Класс com.magnit.flux.mybatis.dao.MyBatisFluxResultProducer

public class MyBatisFluxResultProducer<T> { 
     
   private final SqlSessionFactory sqlSessionFactory; 
 
    private Cursor<T> cursor; 
 
    private SqlSession sqlSession; 
 
    /** 
     * @param cursorFunction функция для продюсирования курсора 
     * @return Flux соответсвующего типа 
     */ 
    public Flux<T> execute(Function<SqlSession, Cursor<T>> cursorFunction) { 
        Optional.ofNullable(sqlSession).ifPresent(e -> { 
            throw new RuntimeException("Cursor already open"); 
        }); 
 
        sqlSession = sqlSessionFactory.openSession(); 
        cursor = cursorFunction.apply(sqlSession); 
        return Flux.fromStream(StreamSupport.stream(cursor.spliterator(), false)); 
    } 
 
    @SneakyThrows 
    public Mono<Void> close() { 
        if (cursor.isOpen()) { 
            cursor.close(); 
 
        } 
        sqlSession.close(); 
        return Mono.empty(); 
    } 
 
} 

Контроллер, работающий с MyBatis 

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

Класс com.magnit.flux.mybatis.controller.MyBatisOperationController

public class MyBatisOperationController { 
     
   private final SqlSessionFactory sqlSessionFactory; 
 
    private final ObjectFactory<MyBatisFluxResultProducer<Operation>> streamResultProducerObjectFactory; 
 
    @GetMapping(path = "/operations-stream", produces = "application/stream+json") 
    public Flux<Operation> getOperationsStream() { 
        return getOperationFlux(); 
    } 
 
    private Flux<Operation> getOperationFlux() { 
        Function<SqlSession, Cursor<Operation>> cursorFunction = sqlSession -> sqlSession 
            .getMapper(OperationMapper.class).getAllOperations(); 
 
        Mono<MyBatisFluxResultProducer<Operation>> streamResultExecutorMono = Mono 
            .just(streamResultProducerObjectFactory.getObject()); 
        return Flux.usingWhen(streamResultExecutorMono, 
            se -> se.execute(cursorFunction), 
            MyBatisFluxResultProducer::close); 
    } 
 
} 

Результат также можно посмотреть в браузере.

Проблема приёма данных на стороне БД СМ

Проблема с передачей больших объёмов данных была решена, но возникла проблема перегрузки БД запросами. Необходимо было ограничить количество запросов на запись в БД. На помощь приходит backpressure (обратное давление).  

Использование backet4j для реализации backpressure 

• Bucket4j — это библиотека ограничения скорости, основанная на алгоритме token-bucket 

• Bucket4j — это потокобезопасная библиотека, которую можно использовать как в автономном приложении JVM, так и в кластерной среде 

• Bucket4j поддерживает кэширование в памяти или распределенное кэширование с помощью спецификации JCache (JSR107) 

Алгоритм Token Bucket

• Создаётся корзина, вместимость которого определяется количеством токенов, которое она может вместить 

• Всякий раз, когда потребитель хочет получить доступ к конечной точке API, он должен получить токен из корзины 

• Токен из корзины удаляется, если он доступен, и запрос принимается 

• Если в корзине нет токенов, то запрос отклоняется 

• Токены пополняются с фиксированной скоростью, которая и является ограничением 

Подробнее о bucket4j здесь.

Bucket4j и Flux 

В данной статье представлены два варианта реализации связки Bucket4j и Flux 

• Блокирующая с использованием методов Flux.delayUntil и Bucket.asBlocking 

• Асинхронная, использующая механизм управления подписчиком (Subscription) и Bucket.asScheduler 

Реактивный клиента на java с использованием Bucket4j 

В классе RateLimitService реализованы два соответствующих метода, которые с заданной скоростью фетчат записи из реактивного http-сервиса, который описан выше. 

Использование блокирующего API bucket4j для backpressure 

Класс com.magnit.fluex.client.service.RateLimitService.executeBlocking

public void executeBlocking(long rate) { 
    AtomicLong counter = new AtomicLong(0); 
    val limit = Bandwidth.simple(rate, Duration.ofMinutes(1)); 
    val bucket = Bucket.builder().addLimit(limit).build(); 
    val flux = client.get().retrieve().bodyToFlux(Operation.class); 
    flux.delayUntil(operation -> { 
        try { 
            bucket.asBlocking().consume(1); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
        return Mono.just("two"); 
    }).subscribe(value ->  System.out.println(counter.incrementAndGet() + " " + value)); 
} 

Основу логики составляет метод Flux.delayUntil в связке bucket.asBlocking.  

При вызове bucket.asBlocking токен либо возвращается, либо поток блокируется до пополнения корзины. 

Использование шедулинга bucket4j для backpressure 

Класс com.magnit.flux.client.service.RateLimitService.executeScheduling

public void executeScheduling(long rate) { 
    AtomicLong counter = new AtomicLong(0); 
    val service = Executors.newSingleThreadScheduledExecutor(); 
    val limit = Bandwidth.simple(rate, Duration.ofMinutes(1)); 
    val bucket = Bucket.builder().addLimit(limit).build(); 
    final long watTime = 50; 
    BaseSubscriber<Operation> subscriber = new BaseSubscriber<Operation>() { 
        private void handleFutureResult(Boolean result) { 
            if (result) { 
                this.request(1); 
            } else { 
                service.schedule(() -> asyncRequest(), watTime, TimeUnit.NANOSECONDS); 
            } 
        } 
 
        @SneakyThrows 
        private void asyncRequest() { 
            val future = bucket.asScheduler().tryConsume(50, watTime, service); 
            if (future.isDone()) { 
                handleFutureResult(future.get()); 
            } else { 
                future.thenAcceptAsync(value -> { 
                    handleFutureResult(value); 
                }); 
            } 
        } 
 
        @Override 
        protected void hookOnSubscribe(Subscription subscription) { 
            asyncRequest(); 
        } 
 
        @Override 
        protected void hookOnNext(Operation value) { 
            System.out.println(counter.incrementAndGet() + " " + value); 
            asyncRequest(); 
        } 
    }; 
    val flux = client.get().retrieve().bodyToFlux(Operation.class); 
    flux.subscribe(subscriber); 
} 

В данном методе основной поток не блокируется. Создаётся подписчик, наследованный от класса из библиотеки reactor — BaseSubscriber. Токен получается методом bucket.asScheduler, который возвращает объект CompletableFuture. Если токен удалось получить (CompletableFuture .isDone() == true), тогда фетчится очередная запись, иначе фетч запускается после пополнения корзины. Ожидание пополнения происходит асинхронно — метод CompletableFuture.thenAcceptAsync. 

Выводы

• В сервисе нам удалось совместить блокирующую и реактивную архитектуры 

• В выборе между hibernate и mybatis мы остановились на hibernate, так как, во-первых, приложение было уже написано с использованием hibernate, а во-вторых, нам важно кеширование данных. 

• Помимо реактивности, мы получили возможность использовать в приложении весь богатый функционал flux.