DefaultReactiveRepository extends DefaultRepository<E> implements ReactiveRepository<E, K> { @Override public Mono<Integer> insertBatch(Publisher<? extends Collection<E>> data) { return Flux.from(data) .filter(CollectionUtils::isNotEmpty) .flatMap(e -> doInsert(e).reactive().flux()) .reduce(Math::addExact) .defaultIfEmpty(0); } DefaultReactiveRepository(DatabaseOperator operator, String table, Class<E> type, ResultWrapper<E, ?> wrapper);  DefaultReactiveRepository(DatabaseOperator operator, RDBTableMetadata table, Class<E> type, ResultWrapper<E, ?> wrapper);  DefaultReactiveRepository(DatabaseOperator operator, Supplier<RDBTableMetadata> table, Class<E> type, ResultWrapper<E, ?> wrapper); @Override Mono<E> newInstance(); @Override Mono<E> findById(Mono<K> primaryKey); @Override Flux<E> findById(Flux<K> key); @Override Mono<Integer> deleteById(Publisher<K> key); @Override Mono<Integer> updateById(K id, Mono<E> data); @Override Mono<SaveResult> save(Publisher<E> data); @Override Mono<Integer> insert(Publisher<E> data); @Override Mono<Integer> insertBatch(Publisher<? extends Collection<E>> data); @Override ReactiveQuery<E> createQuery(); @Override ReactiveUpdate<E> createUpdate(); @Override ReactiveDelete createDelete();  }
@Test public void testInsertBatchReactive() { List<TestEntity> list = Flux.range(0, 100) .map(i -> { TestEntity entity = new TestEntity(); entity.setId("test_" + i); entity.setName("test_" + i); if (i < 10) { entity.setState((byte) 1); } entity.setCreateTime(new Date()); return entity; }) .collectList() .block(); Assert.assertEquals(Integer.valueOf(100), reactiveRepository.insertBatch(Mono.just(list)).block()); }