Spring Boot MongoDB Reactor 教程展示了如何在 Spring Boot 应用中使用 MongoDB 进行反应式编程。
MongoDB
MongoDB 是 NoSQL 跨平台的面向文档的数据库。 它是可用的最受欢迎的数据库之一。 MongoDB 由 MongoDB Inc.开发，并作为免费和开源软件发布。
Spring Data MongoDB 项目提供了与 MongoDB 文档数据库的集成。
反应式编程
反应式编程是一种编程范例，它是功能性的，基于事件的，非阻塞的，异步的，并且以数据流处理为中心。 术语反应式来自以下事实：我们对诸如鼠标单击或 I / O 事件之类的更改做出反应。
当我们处理大量流数据时，响应式应用可以更好地扩展，并且效率更高。 反应性应用是非阻塞的； 他们没有使用资源等待流程完成。
在构建反应式应用时，我们需要它一直到整个数据库都是反应式的。 我们需要使用支持反应式编程的数据库。 MongoDB 是具有响应式支持的数据库。
响应式应用实现基于事件的模型，在该模型中将数据推送到使用者。 数据的使用者称为订阅者，因为它订阅了发布者，后者发布异步数据流。
Spring Reactor
Spring Reactor 是一个反应式库，用于根据反应式流规范在 JVM 上构建非阻塞应用。
Reactor 项目提供两种类型的发布者：Mono和Flux。 Flux是产生 0 到 N 个值的发布者。 返回多个元素的操作使用此类型。 Mono是产生 0 到 1 值的发布者。 它用于返回单个元素的操作。
Spring Boot MongoDB Reactor 示例
在以下应用中，我们对 MongoDB 数据库使用反应式编程。
Note: by default, without any specific configuration, Spring Boot attempts to connect to a locally hosted instance of MongoDB, using the test database name.
pom.xmlsrc├───main│ ├───java│ │ └───com│ │ └───zetcode│ │ │ Application.java│ │ │ MyRunner.java│ │ ├───model│ │ │ City.java│ │ ├───repository│ │ │ CityRepository.java│ │ └───service│ │ CityService.java│ │ ICityService.java│ └───resources│ application.properties└───test └───java
这是 Spring 应用的项目结构。
pom.xml
<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.zetcode</groupId> <artifactId>springbootmongodbreactive</artifactId> <version>1.0-SNAPSHOT</version> <packaging>jar</packaging> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <maven.compiler.source>11</maven.compiler.source> <maven.compiler.target>11</maven.compiler.target> </properties> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.1.5.RELEASE</version> </parent> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build></project>
这是 Maven pom.xml文件。 spring-boot-starter-data-mongodb-reactive是用于使用 MongoDB 面向文档的数据库和 Spring Data MongoDB Reactive 的 Spring Boot 入门程序。
resources/application.properties
spring.main.banner-mode=off
在application.properties中，我们关闭 Spring Boot 标语并设置日志记录属性。 默认情况下，Spring Boot 会尝试使用测试数据库连接到 MongoDB 的本地托管实例。
# mongodbspring.data.mongodb.host=localhostspring.data.mongodb.port=27017spring.data.mongodb.database=testdb
如果要配置 MongoDB，可以设置相应的属性。
com/zetcode/model/City.java
package com.zetcode.model;import org.springframework.data.annotation.Id;import org.springframework.data.mongodb.core.mapping.Document;import java.util.Objects;@Document(value="cities")public class City { @Id private String id; private String name; private int population; public City() { } public City(String name, int population) { this.name = name; this.population = population; } public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public int getPopulation() { return population; } public void setPopulation(int population) { this.population = population; } @Override public int hashCode() { int hash = 7; hash = 79 * hash + Objects.hashCode(this.id); hash = 79 * hash + Objects.hashCode(this.name); hash = 79 * hash + this.population; return hash; } @Override public boolean equals(Object obj) { if (this == obj) { return true; } if (obj == null) { return false; } if (getClass() != obj.getClass()) { return false; } final City other = (City) obj; if (this.population != other.population) { return false; } if (!Objects.equals(this.name, other.name)) { return false; } return Objects.equals(this.id, other.id); } @Override public String toString() { var builder = new StringBuilder(); builder.append("City{id=").append(id).append(", name=") .append(name).append(", population=") .append(population).append("}"); return builder.toString(); }}这是City bean，具有三个属性：id，name和population。
@Document(value="cities")public class City {Bean 用可选的@Document注解修饰。
@Idprivate String id;
id用@Id注解修饰。 Spring 会为一个新生成的城市对象自动生成一个新的 id。
com/zetcode/repository/CityRepository.java
package com.zetcode.repository;import com.zetcode.model.City;import org.springframework.context.annotation.Configuration;import org.springframework.data.mongodb.repository.ReactiveMongoRepository;@Configurationpublic interface CityRepository extends ReactiveMongoRepository<City, String> {}通过从ReactiveMongoRepository扩展，我们有了一个反应性 MongoDB 存储库。
com/zetcode/service/ICityService.java
package com.zetcode.service;import com.zetcode.model.City;import reactor.core.publisher.Flux;import reactor.core.publisher.Mono;import java.util.List;public interface ICityService { Mono<City> insert(City city); Flux<City> saveAll(List<City> cities); Mono<City> findById(String id); Flux<City> findAll(); Mono<Void> deleteAll();}ICityService包含五种合同方法。
com/zetcode/MyRunner.java
package com.zetcode;import com.zetcode.model.City;import com.zetcode.service.CityService;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.CommandLineRunner;import org.springframework.stereotype.Component;import reactor.core.publisher.Flux;import reactor.core.publisher.Mono;import java.util.List;@Componentpublic class MyRunner implements CommandLineRunner { private static final Logger logger = LoggerFactory.getLogger(MyRunner.class); @Autowired private CityService cityService; @Override public void run(String... args) throws Exception { logger.info("Creating cities"); var cities = List.of(new City("Bratislava", 432000), new City("Budapest", 1759000), new City("Prague", 1280000), new City("Warsaw", 1748000)); Mono<Void> one = cityService.deleteAll(); Flux<City> two = cityService.saveAll(cities); Flux<City> three = cityService.findAll(); three.subscribe(city -> logger.info("{}", city)); Mono<Void> all = Mono.when(one, two, three); all.block(); }}我们有一个命令行运行器。 在其run()方法中，我们使用反应式编程访问 MongoDB。
Mono<Void> one = cityService.deleteAll();
如果集合中有城市，我们将删除所有城市。
Flux<City> two = cityService.saveAll(cities);
我们保存城市列表。
Flux<City> three = cityService.findAll();three.subscribe(System.out::println);
我们从集合中找到所有城市。 我们使用subscribe()方法订阅发布者，并将检索到的城市打印到终端。
Mono<Void> all = Mono.when(one, two, three);
使用Mono.when()，我们将三个发布者汇总到一个新的 Mono 中，当所有来源完成后，这些 Mono 将实现。
all.block();
使用block()，我们将触发所有三个操作并等待完成。 由于我们具有控制台应用，因此我们引入了阻塞操作，以便在终端上获得结果。
subscribe()方法开始工作并立即返回。 我们不能保证在应用的其他部分运行操作完成。 block()是一项阻止操作：它触发该操作并等待其完成。
Note: Generally, we rarely use blocking calls in our applications. Operations in a console application is one of a few exceptions.
com/zetcode/Application.java
package com.zetcode;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); }}这段代码设置了 Spring Boot 应用。
在本教程中，我们学习了如何在 Spring Boot 应用中使用反应式编程模型对 MongoDB 进行编程。
