实战SpringCloud响应式微服务系列教程（第十章）响应式RESTful服务完整代码示例

server:
  port: 9801
spring:
  application:
    name: advert
  data:
    mongodb:
      uri: mongodb://localhost:27017/db_advert
  http:
    encoding:
      force: true
      charset: UTF-8
      enabled: true
  redis:
    host: 127.0.0.1
    password: 123456
logback:
  level: info

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

/**
 * 广告投放
*/
@Document(collection="advert")//集合名
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class Advert implements Serializable {
    private static final long serialVersionUID = -8985545025018238754L;
    /**
     * 主键
     */
    @Id
    private String id;
    /**
     * 内容
     */
    private String content;
    /**
     * 发布人
     */
    private Long userId;
    /**
     * 创建时间
     */
    private Date creatData;
    /**
     * 图片地址
     */
    private String imgUrl;
    /**
     * 视频地址
     */
    private String videoUrl;
    /**
     * 广告类型（视频图片）
     */
    private String advertType;
    /**
     * 今日投放地区
     */
    private String launchArea;
    /**
     * 投放时长(小时为单位)
     */
    private int durationTime;
    /**
     * 广告类型
     */
    private int classify;
    /**
     * 计费方式
     */
    private int billingMode;
    /**
     * 展示位置（首页轮播，其他轮播，首页其他位置，其他）
     */
    private int displayPosition;
    /**
     * 广告主题
     */
    private String advertTitle;
    /**
     * 是否需要自定义展示页面
     */
    private int isCustom;
    /**
     * 索引关键词
     */
    private String keyWords;
}

import com.shmc.advert.model.po.Advert;
import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
import org.springframework.data.mongodb.repository.Tailable;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
@Repository
public interface AdvertRepository extends ReactiveMongoRepository<Advert,Long> {
    @Tailable
    Flux<Advert> findBy();
}
 
其中@Repository是org.springframework.stereotype.Repository的注解，这个大家应该都很熟悉了不做解释。
从以上代码中我们可以清楚看到AdvertRepository 继承了ReactiveMongoRepository，ReactiveMongoRepository正是我们依赖的maven响应式mongodb-reactive中的类，其中@Tailable注解，该注解类似于Linux中的tail ，可以将DB的变化以响应式流的方式获取到并推送给前端。
除了可以继承ReactiveMongoRepository之外我们还可以通过注入MongoTemplate来操作mongodb，但是MongoTemplate做不到实时监控和主动推送。如：
@Autowired
MongoTemplate mongoTemplate;

Service接口：
import com.shmc.advert.model.po.Advert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public interface AdvertService {
    Mono<Advert> saveAdvert(Advert advert);
    Mono<Advert> findById(String id);
    Flux<Advert> findAll();
    Flux<Advert> findByAll();
}
 
Service实现类：
@Service
public class AdvertServiceImpl implements AdvertService {
    @Autowired
    MongoTemplate mongoTemplate;
    @Autowired
    private AdvertRepository advertRepository;
    @Override
    public Mono<Advert> saveAdvert(Advert advert){
        advert.setId(new IdWorker().nextId());
        advert.setCreatData(new Date());
        mongoTemplate.insert(advert);
        return Mono.just(advert);
    }
    @Override
    public Mono<Advert> findById(String id){
        Query query = new Query(Criteria.where("id").is(id));
        return Mono.just(mongoTemplate.findOne(query,Advert.class));
    }
    @Override
    public Flux<Advert> findAll(){
        return advertRepository.findAll();
    }
    @Override
    public Flux<Advert> findByAll(){
        return advertRepository.findBy();
    }
}

import com.shmc.advert.model.po.Advert;
import com.shmc.advert.service.AdvertService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
@RestController
@RequestMapping("/advert")
public class AdvertController {
    @Autowired
    private AdvertService advertService;
    @PostMapping("/saveAdvert")
    public Mono<Advert> saveAdvert(@RequestBody Advert advert){
        return advertService.saveAdvert(advert);
    }
    @GetMapping("/findById/{id}")
    public Mono<Advert> findById(@PathVariable String id){
        return advertService.findById(id);
    }
    /**
     * 以stream+json流的方式推送到客户端
     * @return
     */
    @GetMapping(value = "/findAllPreSec", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<Advert> findAllPreSec() {
        return advertService.findAll().delayElements(Duration.ofSeconds(1));
    }
    /**
     * 数据变更
     * @return
     */
    @GetMapping(value = "/findByAll", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<Advert> findByAll(){
        return advertService.findByAll();
    }
}