作者:
瑾年微信公众号: Java知音
1.搭建响应式RESTful服务。
在前面章节中我们讲了如何使用 Spring Initializer初始化响应式web应用，本节中就不再做过多介绍
在学习本章内容之前需要了解mongodb以及redis，mongodb以及redis可查阅相关资料进行全面了解，并在本地环境搭建mongodb和redis。
2.application.yml文件配置
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
以上配置代码是我们目前学习的响应式RESTful服务的全部配置，配置比较简单，spring.data.mongodb.uri: mongodb://localhost:27017/db_advert和spring.data.redis是我们服务的核心配置，我们知道传统的数据库是不支持响应式数据读取的，所以这里使用mongodb和redis代替。
3.集成响应式的MongoDB
springboot 本身提供了cassandra/couchbase/mongodb/redis这几个NoSQL数据库的响应式驱动：
阻塞式的spring-boot-starter-data-mongodb 改为响应式的mongodb依赖spring-boot-starter-data-mongodb-reactive：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>
（1）编写实体类：
/**
* 广告投放
*/
@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;
}
其中@document把一个java类声明为mongodb的文档,可以通过collection参数指定这个类对应的文档,标注在实体类上,类似于hibernate的entity注解。其他注解均为lombok的注解。
（2）编写repository接口
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;
（3）编写Service接口以及实现类
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();
}
}
（4）编写Controller
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();
}
}
至此基于RESTful的响应式服务我们全部完成了，启动程序访问“/advert/findAllPreSec"接口和”/advert/findByAll“接口就可以看到响应式的数据推送了。
findAllPreSec这个方法。使用了delayElements使得每隔一秒钟获取一条数据发送给客户端，以“异步响应式流”的方式逐条推送。
这里指定了MediaType是APPLICATION_STREAM_JSON，即application/stream+json格式。
在浏览器中就可以看到每隔一秒出现一条记录。运行程序，访问findByAll接口，然后测试调用save接口添加advert数据，或者直接通过MongoDB Compass客户端添加Stu数据，就会看到在页面中实时看到新添加的数据了。
下一章会吧代码上传到gitee，各位看官如有需要请自行下载。
