springwebflux使用的详细代码_springwebflux使用资源-CSDN文库

共37个文件

class：19个

java：15个

yaml：2个

版权申诉

142 浏览量 2023-10-26 09:03:28 上传评论收藏 44KB ZIP 举报

Spring WebFlux是Spring Framework 5.0引入的一个全新模块，它是对传统Spring MVC的补充，专注于非阻塞式、反应式编程模型。这个模型特别适用于高并发、低延迟的现代Web应用程序。在这个主题中，我们将深入探讨Spring WebFlux的使用，包括它如何与Spring Boot集成，以及如何通过具体的代码示例来实现各种功能。让我们理解什么是反应式编程。反应式编程是一种编程范式，强调数据流和依赖传播，使得系统能够快速响应变化。在WebFlux中，这体现在异步处理请求和事件驱动的模型上，能够更有效地利用系统资源。 Spring WebFlux的核心组件是`Reactor`，一个遵循 Reactive Streams 规范的库。它提供了一种方式来处理和发布异步事件，使开发者能够构建高效的反应式应用。 **一、Spring Boot与Spring WebFlux的整合** 要将Spring WebFlux与Spring Boot结合，你需要在`build.gradle`或`pom.xml`文件中添加相关依赖。对于Maven，可以添加以下依赖： ```xml <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency> ``` **二、创建WebFlux处理程序** 在Spring WebFlux中，我们使用`@RestController`注解的类和`@RequestMapping`、`@GetMapping`、`@PostMapping`等方法来定义HTTP端点。不过，它们使用的是`Mono`或`Flux`类型，这两个类型分别代表0个或1个，以及0个到多个结果的异步序列。下面是一个简单的GET请求处理器示例： ```java @RestController public class GreetingController { @GetMapping("/greeting") public Mono<String> greeting() { return Mono.just("Hello, Spring WebFlux!"); } } ``` **三、WebFlux配置** Spring Boot自动配置会创建一个默认的WebFlux配置，但你可以自定义配置，比如选择使用Netty、Tomcat或其他服务器。在`application.properties`或`application.yml`中，你可以设置`server.netty.enabled=true`来使用Netty。 **四、数据访问与反应式数据库** WebFlux与反应式数据源（如R2DBC）的集成允许进行非阻塞数据库访问。例如，使用R2DBC访问MySQL： ```java @Configuration public class DatabaseConfig { @Bean public ConnectionFactory connectionFactory() { return DatabaseClient.create(new MysqlConnectionConfiguration.Builder() .host("localhost") .port(3306) .username("root") .password("password") .database("test") .build()); } @Bean public RepositoryFactorySupport repositoryFactory(ConnectionFactory connectionFactory) { return new R2dbcRepositoryFactory(connectionFactory); } } ``` **五、错误处理** WebFlux提供了`HandlerExceptionResolver`接口来处理异常。你可以创建一个自定义实现，捕获并处理可能出现的错误： ```java @Component public class CustomExceptionHandler implements HandlerExceptionResolver { @Override public Mono<Void> resolve(ServerWebExchange exchange, Throwable ex) { // 处理异常逻辑 } } ``` **六、单元测试与端到端测试** Spring Boot为WebFlux提供了`WebTestClient`，可以方便地进行单元测试和端到端测试。例如，测试上面的`GreetingController`： ```java @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) class GreetingControllerTest { @Autowired private WebTestClient webTestClient; @Test void testGreeting() { webTestClient.get().uri("/greeting") .exchange() .expectStatus().isOk() .expectBody(String.class).isEqualTo("Hello, Spring WebFlux!"); } } ``` 以上就是Spring WebFlux的基本使用和关键概念，结合提供的"boot-webflux"项目，你可以进一步探索更多细节，包括路由定义、过滤器、中间件、自定义异常处理、反应式数据访问等功能，从而全面掌握Spring WebFlux在实际项目中的应用。

资源推荐

资源详情

资源评论

收起资源包目录

boot-webflux.zip （37个子文件）

boot-webflux

pom.xml 2KB

src

test

java

main

resources

application.yaml 23B

java

com

congge

handler

ApiHandler.java 1KB

ClientTest.java 1013B

MyRoutesConfig.java 993B

BookHandler.java 2KB

web

BookController.java 974B

FluxApp.java 1KB

service

BookService.java 313B

impl

BookServiceImpl.java 1KB

entity

BookInfo.java 278B

router

BookServer.java 2KB

ApiRouterFunction.java 1KB

reactors

ReactorApi.java 3KB

ReactorStreamApi.java 3KB

StreamTest.java 2KB

ParallelFluxTest.java 1KB

target

classes

application.yaml 23B

com

congge

handler

ApiHandler.class 3KB

MyRoutesConfig.class 3KB

BookHandler.class 4KB

ClientTest.class 4KB

web

BookController.class 2KB

service

BookService.class 622B

impl

BookServiceImpl.class 3KB

entity

BookInfo.class 2KB

FluxApp.class 2KB

router

ApiRouterFunction.class 2KB

BookServer.class 4KB

reactors

ParallelFluxTest$1.class 2KB

ReactorStreamApi$MySubscriber.class 1KB

ReactorApi$MyDataListener.class 1KB

ReactorApi$1.class 1KB

ParallelFluxTest.class 1KB

ReactorStreamApi.class 6KB

ReactorApi.class 7KB

StreamTest.class 5KB

generated-sources

annotations

package com.congge.reactors; import org.junit.Test; import reactor.core.publisher.Flux; import reactor.core.publisher.FluxSink; import java.time.Duration; import java.time.temporal.ChronoUnit; import java.util.Arrays; import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Stream; public class ReactorApi { @Test public void fluxJust() { Flux<String> phones = Flux.just("小米", "三星", "华为"); } @Test public void fluxFromIterable() { Flux<String> phones = Flux.fromIterable(Arrays.asList("小米", "三星", "华为")); } @Test public void fluxFromArray() { Flux<String> phones = Flux.fromArray(new String[]{"小米", "三星", "华为"}); } @Test public void fluxFromStream() { Flux<String> phones = Flux.fromStream(Stream.of(new String[]{"小米", "三星", "华为"})); phones.subscribe(); phones.subscribe(); //只能被订阅一次 } @Test public void fluxEmpty() { Flux<String> phones = Flux.empty(); //generic type still honored } @Test public void fluxRange() { Flux<Integer> phones = Flux.range(5, 3); } @Test public void fluxGenerate() { Flux<Long> flux = Flux.generate( AtomicLong::new, (state, sink) -> { long i = state.getAndIncrement(); sink.next(i); if (i == 10) sink.complete(); return state; }, (state) -> System.out.println("done") ); flux.subscribe(System.out::println); } @Test public void fluxCreate() { Flux<String> phones = Flux.create((t) -> { t.next("小米"); t.next("三星"); t.next("华为"); t.complete(); }); phones.subscribe(System.out::println); System.out.println("------------"); Flux<String> ownFluxListener = Flux.create(sink -> { //传入自定义的方法 new MyDataListener(){ public void onReceiveData(String str){ sink.next(str); } public void onComplete(){ sink.complete(); } }; }, FluxSink.OverflowStrategy.DROP); ownFluxListener.subscribe(System.out::println); } public class MyDataListener{ public void onReceiveData(String str){ System.out.println("收到数据:"+str); } public void onComplete(){ System.out.println("完成数据的消费处理"); } } @Test public void fluxDefer() { Flux.defer(() -> Flux.just("小米", "三星", "华为")) .subscribe(System.out::println); Flux<String> stockSeq4 = Flux.defer(() -> Flux.fromStream(Stream.of(new String[]{"小米", "三星", "华为"}))); stockSeq4.subscribe(); stockSeq4.subscribe(); } @Test public void fluxInterval() throws InterruptedException { //interval 定时发送元素 Flux.interval(Duration.of(1, ChronoUnit.SECONDS)) .subscribe((t) -> System.out.println((String.valueOf(t)))); Thread.sleep(1000000); } }

评论收藏

内容反馈

版权申诉