快捷導(dǎo)航

Spring?Boot?整合?Reactor實(shí)例詳解

更新時(shí)間：2022年09月08日 14:16:16 作者：六七十三

這篇文章主要為大家介紹了Spring?Boot?整合?Reactor實(shí)例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

引言

Reactor 是一個(gè)完全非阻塞的 JVM 響應(yīng)式編程基礎(chǔ)，有著高效的需求管理（背壓的形式）。它直接整合 Java8 的函數(shù)式 API，尤其是 CompletableFuture， Stream，還有 Duration 。提供了可組合的異步化序列 API — Flux (對于 [N] 個(gè)元素) and Mono (對于 [0|1] 元素) — 并廣泛實(shí)現(xiàn) 響應(yīng)式Stream 規(guī)范。

這次帶大家從零開始，使用 Spring Boot 框架建立一個(gè) Reactor 響應(yīng)式項(xiàng)目。

1 創(chuàng)建項(xiàng)目

使用 ;https://start.spring.io/ 創(chuàng)建項(xiàng)目。添加依賴項(xiàng)：H2、Lombok、Spring Web、JPA、JDBC

然后導(dǎo)入 Reactor 包

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <scope>test</scope>
</dependency>

2 集成 H2 數(shù)據(jù)庫

application.properties 文件中添加 H2 數(shù)據(jù)連接信息。此外，端口使用 8081（隨意，本地未被使用的端口即可）。

server.port=8081
################ H2 數(shù)據(jù)庫 基礎(chǔ)配置 ##############
spring.datasource.driverClassName=org.h2.Driver
spring.datasource.url=jdbc:h2:~/user
spring.datasource.username=sa
spring.datasource.password=
spring.jpa.database=h2
spring.jpa.hibernate.ddl-auto=update
spring.h2.console.path=/h2-console
spring.h2.console.enable=true

3 創(chuàng)建測試類

3.1 user 實(shí)體

建立簡單數(shù)據(jù)操作實(shí)體 User。

import lombok.Data;
import lombok.NoArgsConstructor;
import javax.persistence.*;
/**
 * @Author: prepared
 * @Date: 2022/8/29 21:40
 */
@Data
@NoArgsConstructor
@Table(name = "t_user")
@Entity
public class User {
    @Id
    @GeneratedValue(strategy = GenerationType.AUTO)
    private Long id;
    private String userName;
    private int age;
    private String sex;
    public User(String userName, int age, String sex) {
        this.userName = userName;
        this.age = age;
        this.sex = sex;
    }
}

3.2 UserRepository

數(shù)據(jù)模型層使用 JPA 框架。

import com.prepared.user.domain.User;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;
/**
 * @Author: prepared
 * @Date: 2022/8/29 21:45
 */
@Repository
public interface UserRepository extends JpaRepository<User, Long> {
}

3.3 UserService

service 增加兩個(gè)方法，add 方法，用來添加數(shù)據(jù)；list 方法，用來查詢所有數(shù)據(jù)。所有接口返回 Mono/Flux 對象。

最佳實(shí)踐：所有的第三方接口、IO 耗時(shí)比較長的操作都可以放在 Mono 對象中。

doOnError 監(jiān)控異常情況；

doFinally 監(jiān)控整體執(zhí)行情況，如：耗時(shí)、調(diào)用量監(jiān)控等。

import com.prepared.user.dao.UserRepository;
import com.prepared.user.domain.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import javax.annotation.Resource;
import java.util.List;
/**
 * @Author: prepared
 * @Date: 2022/8/29 21:45
 */
@Service
public class UserService {
    private Logger logger = LoggerFactory.getLogger(UserService.class);
    @Resource
    private UserRepository userRepository;
    public Mono<Boolean> save(User user) {
        long startTime = System.currentTimeMillis();
        return Mono.fromSupplier(() -> {
                    return userRepository.save(user) != null;
                })
                .doOnError(e -> {
                    // 打印異常日志&增加監(jiān)控（自行處理）
                    logger.error("save.user.error, user={}, e", user, e);
                })
                .doFinally(e -> {
                    // 耗時(shí) & 整體健康
                    logger.info("save.user.time={}, user={}", user, System.currentTimeMillis() - startTime);
                });
    }
    public Mono<User> findById(Long id) {
        long startTime = System.currentTimeMillis();
        return Mono.fromSupplier(() -> {
                    return userRepository.getReferenceById(id);
                }).doOnError(e -> {
                    // 打印異常日志&增加監(jiān)控（自行處理）
                    logger.error("findById.user.error, id={}, e", id, e);
                })
                .doFinally(e -> {
                    // 耗時(shí) & 整體健康
                    logger.info("findById.user.time={}, id={}", id, System.currentTimeMillis() - startTime);
                });
    }
    public Mono<List<User>> list() {
        long startTime = System.currentTimeMillis();
        return Mono.fromSupplier(() -> {
                    return userRepository.findAll();
                }).doOnError(e -> {
                    // 打印異常日志&增加監(jiān)控（自行處理）
                    logger.error("list.user.error, e", e);
                })
                .doFinally(e -> {
                    // 耗時(shí) & 整體健康
                    logger.info("list.user.time={}, ", System.currentTimeMillis() - startTime);
                });
    }
  public Flux<User> listFlux() {
        long startTime = System.currentTimeMillis();
        return Flux.fromIterable(userRepository.findAll())
                .doOnError(e -> {
                    // 打印異常日志&增加監(jiān)控（自行處理）
                    logger.error("list.user.error, e", e);
                })
                .doFinally(e -> {
                    // 耗時(shí) & 整體健康
                    logger.info("list.user.time={}, ", System.currentTimeMillis() - startTime);
                });
    }
}

3.4 UserController

controller 增加兩個(gè)方法，add 方法，用來添加數(shù)據(jù)；list 方法，用來查詢所有數(shù)據(jù)。

list 方法還有另外一種寫法，這就涉及到 Mono 和 Flux 的不同了。

返回List可以使用Mono<List<User>> ，也可以使用 Flux<User>。

Mono<T> 是一個(gè)特定的 Publisher<T>，最多可以發(fā)出一個(gè)元素
Flux<T> 是一個(gè)標(biāo)準(zhǔn)的 Publisher<T>，表示為發(fā)出 0 到 N 個(gè)元素的異步序列

import com.prepared.user.domain.User;
import com.prepared.user.service.UserService;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
/**
 * @Author: prepared
 * @Date: 2022/8/29 21:47
 */
@RestController
public class UserController {
    @Resource
    private UserService userService;
    @RequestMapping("/add")
    public Mono<Boolean> add() {
        User user = new User("xiaoming", 10, "F");
        return userService.save(user) ;
    }
    @RequestMapping("/list")
    public Mono<List<User>> list() {
        return userService.list();
    }
}
    @RequestMapping("/listFlux")
    public Flux<User> listFlux() {
        return userService.listFlux();
    }

3.5 SpringReactorApplication 添加注解支持

Application 啟動(dòng)類添加注解 @EnableJpaRepositories

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
/**
 * Hello world!
 */
@SpringBootApplication
@EnableJpaRepositories
public class SpringReactorApplication {
    public static void main(String[] args) {
        SpringApplication.run(SpringReactorApplication.class, args);
    }
}

測試

啟動(dòng)項(xiàng)目，訪問 localhost:8081/add，正常返回 true。

查詢所有數(shù)據(jù)，訪問localhost:8081/list，可以看到插入的數(shù)據(jù)，已經(jīng)查詢出來了。PS：我這里執(zhí)行了多次 add，所以有多條記錄。

后臺日志：

2022-09-05 20:13:17.385 INFO 15696 --- [nio-8082-exec-2] com.prepared.user.service.UserService : list.user.time=181,

執(zhí)行了 UserService list() 方法的 doFinnally 代碼塊，打印耗時(shí)日志。

總結(jié)

響應(yīng)式編程的優(yōu)勢是不會阻塞。那么正常我們的代碼中有哪些阻塞的操作呢？

Future 的 get() 方法;
Reactor 中的 block() 方法，subcribe() 方法，所以在使用 Reactor 的時(shí)候，除非編寫測試代碼，否則不要直接調(diào)用以上兩個(gè)方法；
同步方法調(diào)用，所以高并發(fā)情況下，會使用異步調(diào)用（如Future）來提升響應(yīng)速度

以上就是Spring Boot 整合 Reactor實(shí)例詳解的詳細(xì)內(nèi)容，更多關(guān)于Spring Boot 整合 Reactor的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: