引入maven文件

<dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>

创建配置文件

spring:
  data:
    mongodb:
      host: 127.0.0.1
      port: 27017
      username: mickey
      password: 123456
      database: mongoTest

创建document实体类

@Data
@Document("person")
public class PersonEntity {

    @Id
    private String id;

    private String userName;

    private String gender;

    /**
     * 设置TTL，单位秒
     */
    @Indexed(name = "idx_create_time", expireAfterSeconds = 10)
    private Date createTime = new Date();

}

创建repository

@Repository
public interface PersonRepository extends ReactiveMongoRepository<PersonEntity,String> {

    /**
     * 根据name查找Person
     * @param name
     * @return
     */
    Flux<PersonEntity> findByUserName(String name);
}

编写controller

package com.xzsx.openapi.thirdparty.controller;

import com.xzsx.openapi.dto.MongoDBOutput;
import com.xzsx.openapi.thirdparty.entity.PersonEntity;
import com.xzsx.openapi.thirdparty.repository.PersonRepository;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * @author J·K
 * @Description: TODO
 * @date 2019-04-22 10:19
 */
@RestController
public class IndexController
{
    /**
     * 可以使用
     */
    @Autowired
    private PersonRepository personRepository;

    @GetMapping("/save")
    public Mono<PersonEntity> save(){
        PersonEntity person = new PersonEntity();
        person.setUserName("mickey");
        person.setGender("male");
        return personRepository.save(person);
    }

    @GetMapping("/list")
    public Flux<MongoDBOutput> list(){
        Flux<MongoDBOutput> flux = personRepository.findAll().map(x->{
            MongoDBOutput mongoDBOutput = new MongoDBOutput();
            BeanUtils.copyProperties(x,mongoDBOutput);
            return mongoDBOutput;
        });
        return flux;
    }

    @GetMapping("/delete/{id}")
    public Mono<String> delete(@PathVariable("id") String id){
        //没有返回值
//        personRepository.deleteById(id);

        //如果要操作数据，并返回一个Mono，这时候使用flatMap
        //如果不操作数据，只是对数据做一个转换，使用map
        return personRepository.findById(id)
        .flatMap(x->
                personRepository.deleteById(x.getId())
                .then(Mono.just("ok")))
                .defaultIfEmpty("not found");
    }

    @GetMapping("/update/{id}")
    public Mono<String> update(@PathVariable("id") String id){
        return personRepository.findById(id)
                .flatMap(x->{
                    x.setUserName("jack");
                    return personRepository.save(x);
                })
                .map(x->x.toString())
                .defaultIfEmpty("error");
    }

    @GetMapping("/find/{id}")
    public Mono<PersonEntity> findById(@PathVariable("id") String id){
        return personRepository.findById(id)
                .map(x-> x)
                .defaultIfEmpty(new PersonEntity());
    }

    @GetMapping("/findByName/{name}")
    public Flux<PersonEntity> findByName(@PathVariable("name") String name){
        return personRepository.findByUserName(name)
                .map(x-> x)
                .defaultIfEmpty(new PersonEntity());
    }

    @GetMapping(value = "/stream/findByName/{name}",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<PersonEntity> findByName1(@PathVariable("name") String name){
        return personRepository.findByUserName(name)
                .map(x-> x)
                .defaultIfEmpty(new PersonEntity());
    }

    public static void main(String[] args) {
        String [] strs = {"1","2","3"};
//        List<Integer> collect = Flux.fromArray(strs).map(x -> Integer.parseInt(x))
//                .toStream().collect(Collectors.toList());
//        System.out.println(collect);


//        Flux.fromArray(strs).map(x->Integer.parseInt(x))
//        .subscribe(x->{
//            System.out.println(x);
//        });

        Flux.fromArray(strs).map(x->{
            throw new RuntimeException("error");
        })
        .subscribe(x->{
            System.out.println(x);
        },x->{
            System.out.println("error");
        });

    }

}

启动项目后，就可以按原来访问springboot的方式访问数据了