搜索到
51
篇与
的结果
-
响应式编程学习笔记 响应式编程1、Reactor核心前置知识1、Lambda2、Function根据出参，入参分类1、有入参，有出参 --> FunctionFunction<String, Integer> function = a -> Integer.parseInt(a);2、有入参，无出参Consumer<String> consumer = a -> System.out.println(a);3、无入参，有出参Supplier<String> supplier = () -> UUID.randomUUID().toString();4、无入参，无出参Runnable runnable = () -> System.out.println("xixi"); 3、StreamAPI流式操作，三大步骤1、创建流Stream<Integer> integerStream = Stream.of(1, 2, 3); Stream<Integer> stream = list.stream();2、中间操作(intermediate operation)，可以有多个filter,map,mapToInt,mapToLong,mapToDouble,flatMap,flatMapToInt,flatMapToLong,flatMapToDouble,mapMulti,mapMultiToInt,mapMultiToLong,mapMultiToDouble,peek...3、终止操作(terminal operation)，只能有一个forEach,forEachOrdered,toArray,toArray,reduce,collect,toList,min,max,count,anyMatch,findFirst,findAny...流式操作是否并发？ // 流的三大部份 // 1.创建流 2.N个中间操作 3.一个终止操作 Stream<Integer> integerStream = Stream.of(1, 2, 3); Stream<Object> buildStream = Stream.builder().add(1).add(2).add(3).build(); Stream<Object> concatStream = Stream.concat(integerStream, buildStream); Stream<Integer> stream = list.stream(); List<Integer> resultList = new ArrayList<>(); System.out.println("main线程： "+Thread.currentThread().getName()); // 流是不是并发操作？ 答：默认单线程，可以通过parallel开启多线程，但是如果开启多线程，则需要自身注意线程安全问题 long count = list.stream() .parallel() // 开启多线程 并发流 .filter(i -> { // resultList.add(i); // 开启多线程，不能这样写，要保证流里面的数据是无状态的，即流里面的数据只在流内部使用 // 可以计算完成以后返回出去，但是不能在内部又引用外部的数据，可能会出现问题 System.out.println("filter线程: " + Thread.currentThread().getName()); return i > 2; }) .count(); System.out.println(resultList);注意： 要保证流里面的数据是无状态的中间操作：filter：过滤，挑出我们要的元素takeWhile示例List<Integer> collect = Stream.of(1, 2, 3, 4, 5, 6) .filter(a -> a > 2) // 无条件遍历 .toList(); System.out.println(collect); List<Integer> collect1 = Stream.of(1, 2, 3, 4, 5, 6) .takeWhile(a -> a < 2) // 当条件不满足时，直接返回 .toList(); System.out.println(collect1);map：映射，一对一映射mapToInt，MapToDouble..flatMap: 打散、散列、展开，一对多映射...终止操作：forEach、forEachOrdered、toArray、reduce、collect、toList、min、 max、count、anyMatch、allMatch、noneMatch、findFirst、findAny、iterator4、Reactive Stream目的：通过全异步的方式，加缓冲区构建一个实时的数据流系统。kafka，mq能构建大型的分布式响应系统，缺少本地化分布式响应系统方案jvm推出Reactive Stream，让所有异步线程能够互相监听消息，处理消息，构建实时消息处理流Api Component：1、Publisher：发布者2、Subscriber：订阅者3、Processor：处理器响应式编程总结：1、底层：基于数据缓冲队列+消息驱动模型+异步回调机制2、编码：流式编程+链式调用+生命式API3、效果：优雅全异步+消息实时处理+高吞吐量+占用少量资源与传统写法对比：传统写法痛点：以前要做一个高并发系统：缓存、异步、队列，手动控制整个逻辑现在：全自动控制整个逻辑Reactor1、快速上手介绍Reactor 是一个用于JVM的完全非阻塞的响应式编程框架，具备高效的需求管理（即对 “背压（backpressure）”的控制）能力。它与 Java 8 函数式 API 直接集成，比如 CompletableFuture， Stream， 以及 Duration。它提供了异步序列 API Flux（用于[N]个元素）和 Mono（用于 [0|1]个元素），并完全遵循和实现了“响应式扩展规范”（Reactive Extensions Specification）。Reactor 的 reactor-ipc 组件还支持非阻塞的进程间通信（inter-process communication, IPC）。 Reactor IPC 为 HTTP（包括 Websockets）、TCP 和 UDP 提供了支持背压的网络引擎，从而适合 应用于微服务架构。并且完整支持响应式编解码（reactive encoding and decoding）。依赖<dependencyManagement> <dependencies> <dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-bom</artifactId> <version>2023.0.0</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement><dependencies> <dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-core</artifactId> </dependency> <dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-test</artifactId> <scope>test</scope> </dependency> </dependencies>2、响应式编程响应式编程是一种关注于数据流（data streams）和变化传递（propagation of change）的异步编程方式。 这意味着它可以用既有的编程语言表达静态（如数组）或动态（如事件源）的数据流。3、核心特性1、Mono和FluxMono: 0|1 数据流Flux: N数据流响应式流：元素（内容） + 信号（完成/异常）；2、subscribe()自定义流的信号感知回调.subscribe( System.out::println // 消费方法 , throwable -> System.out.println(throwable.getMessage()) // 感知异常 , () -> System.out.println("complete") // 感知正常结束 ); // 流只有被订阅了才会执行，否则没有任何操作自定义消费者.subscribe(new BaseSubscriber<String>() { // 自定义消费者 @Override protected void hookOnSubscribe(Subscription subscription) { System.out.println("被订阅"); requestUnbounded(); } @Override protected void hookOnNext(String value) { System.out.println("下个元素"); } @Override protected void hookOnComplete() { System.out.println("完成信号"); } @Override protected void hookOnError(Throwable throwable) { System.out.println("异常信号"); } @Override protected void hookOnCancel() { System.out.println("结束信号"); } @Override protected void hookFinally(SignalType type) { System.out.println("终止信号"); } });3、流的取消消费者调用 cancle() 取消流的订阅；4、自定义消费者推荐直接编写jdk自带的BaseSubscriber的实现类5、背压（back pressure）和请求重塑（reshape requests）buffer/** * 缓冲区 */ private static void bufferTest() { Flux.range(1, 10).buffer(3).subscribe(v -> System.out.println("v的类型：" + v.getClass() + "的值：" + v)); }limitRate/** * 测试limitRate */ private static void limitTest() { Flux.range(1,1000) .log() .limitRate(100) // 一次预取100个元素 75%预取策略，第一次取100个如果75%已经处理，继续请求新的75%数据 .subscribe(System.out::println); }6、以编程方式创建序列-SinkSink.nextSink.complete1、同步环境-generate/** * 通过generate创建序列 */ private static void generateTest() { List<Integer> list = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9); Flux.generate(() -> 0, // 初始值 (i, a) -> { a.next(list.get(i)); // 把元素放入通道 if (i == list.size() - 1) { a.complete(); // 完成 } return ++i; // 下次回调的元素 } ) .subscribe(System.out::println); }2、多线程-create/** * 通过create创建序列，create适用与多线程环境，generate适用于单线程环境 */ private static void createTest() { Flux.create(sink -> { for (int i = 0; i < 10; i++) { sink.next("2"); } }).subscribe(System.out::println); }7、handle自定义流中的处理规则/** * handle自定义处理 */ private static void handleTest() { Flux.range(1, 10) .handle((value,sink) -> { System.out.println("接收到value:" + value); sink.next("haha_" + value); }) .subscribe(); }8、自定义线程调度响应式：响应式编程： 全异步、消息、事件回调默认还是用当前线程，生成整个流、发布流、流操作/** * 自定义线程测试 */ private static void threadTest() { // 响应式编程：全异步，消息，回调机制 Schedulers.boundedElastic(); // 有界的，弹性线程池 Schedulers.single(); // 单线程 Schedulers.immediate(); // 都在同一个当前线程(默认) Scheduler scheduler = Schedulers.newParallel("my-parallel"); Flux<Integer> flux = Flux.range(1, 10) .publishOn(scheduler) .log(); flux.subscribe(); }9、异常处理命令式编程：常见的错误处理方式Catch and return a static default value. 捕获异常返回一个静态默认值try { return doSomethingDangerous(10); } catch (Throwable error) { return "RECOVERED"; }onErrorReturn: 实现上面效果，错误的时候返回一个值●1、吃掉异常，消费者无异常感知●2、返回一个兜底默认值●3、流正常完成；Catch and execute an alternative path with a fallback method.吃掉异常，执行一个兜底方法；try { return doSomethingDangerous(10); } catch (Throwable error) { return doOtherthing(10); }onErrorResume●1、吃掉异常，消费者无异常感知●2、调用一个兜底方法●3、流正常完成Flux.just(1, 2, 0, 4) .map(i -> "100 / " + i + " = " + (100 / i)).onErrorResume(err -> Mono.just("哈哈-777")) .subscribe(v -> System.out.println("v = " + v), err -> System.out.println("err = " + err), () -> System.out.println("流结束"));Catch and dynamically compute a fallback value. 捕获并动态计算一个返回值根据错误返回一个新值try { Value v = erroringMethod(); return MyWrapper.fromValue(v); } catch (Throwable error) { return MyWrapper.fromError(error); }.onErrorResume(err -> Flux.error(new BusinessException(err.getMessage()+"：炸了")))●1、吃掉异常，消费者有感知●2、调用一个自定义方法●3、流异常完成Catch, wrap to a BusinessException, and re-throw.捕获并包装成一个业务异常，并重新抛出try { return callExternalService(k); } catch (Throwable error) { throw new BusinessException("oops, SLA exceeded", error); }包装重新抛出异常: 推荐用 .onErrorMap●1、吃掉异常，消费者有感知●2、抛新异常●3、流异常完成.onErrorResume(err -> Flux.error(new BusinessException(err.getMessage()+"：炸了"))) Flux.just(1, 2, 0, 4) .map(i -> "100 / " + i + " = " + (100 / i)) .onErrorMap(err-> new BusinessException(err.getMessage()+": 又炸了...")) .subscribe(v -> System.out.println("v = " + v), err -> System.out.println("err = " + err), () -> System.out.println("流结束"));Catch, log an error-specific message, and re-throw.捕获异常，记录特殊的错误日志，重新抛出try { return callExternalService(k); } catch (RuntimeException error) { //make a record of the error log("uh oh, falling back, service failed for key " + k); throw error; }Flux.just(1, 2, 0, 4) .map(i -> "100 / " + i + " = " + (100 / i)) .doOnError(err -> { System.out.println("err已被记录 = " + err); }).subscribe(v -> System.out.println("v = " + v), err -> System.out.println("err = " + err), () -> System.out.println("流结束"));●异常被捕获、做自己的事情●不影响异常继续顺着流水线传播●1、不吃掉异常，只在异常发生的时候做一件事，消费者有感知Use the finally block to clean up resources or a Java 7 “try-with-resource” construct. Flux.just(1, 2, 3, 4) .map(i -> "100 / " + i + " = " + (100 / i)) .doOnError(err -> { System.out.println("err已被记录 = " + err); }) .doFinally(signalType -> { System.out.println("流信号："+signalType); })忽略当前异常，仅通知记录，继续推进Flux.just(1,2,3,0,5) .map(i->10/i) .onErrorContinue((err,val)->{ System.out.println("err = " + err); System.out.println("val = " + val); System.out.println("发现"+val+"有问题了，继续执行其他的，我会记录这个问题"); }) //发生 .subscribe(v-> System.out.println("v = " + v), err-> System.out.println("err = " + err));10、常用操作filter、flatMap、concatMap、flatMapMany、transform、defaultIfEmpty、switchIfEmpty、concat、concatWith、merge、mergeWith、mergeSequential、zip、zipWith...2、Spring Webflux0、组件对比API功能Servlet-阻塞式WebWebFlux-响应式Web前端控制器DispatcherServletDispatcherHandler处理器ControllerWebHandler/Controller请求、响应ServletRequest、ServletResponseServerWebExchange：ServerHttpRequest、ServerHttpResponse过滤器Filter（HttpFilter）WebFilter异常处理器HandlerExceptionResolverDispatchExceptionHandlerWeb配置@EnableWebMvc@EnableWebFlux自定义配置WebMvcConfigurerWebFluxConfigurer返回结果任意Mono、Flux、任意发送REST请求RestTemplateWebClientMono： 返回0|1 数据流Flux：返回N数据流1、WebFlux底层基于Netty实现的Web容器与请求/响应处理机制参照：https://docs.spring.io/spring-framework/reference/6.0/web/webflux.html2、引入<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>3.1.6</version> </parent> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency> </dependencies>Context 响应式上下文数据传递； 由下游传播给上游；以前： 浏览器 --> Controller --> Service --> Dao： 阻塞式编程现在： Dao（数据源查询对象【数据发布者】） --> Service --> Controller --> 浏览器： 响应式大数据流程： 从一个数据源拿到大量数据进行分析计算；ProductVistorDao.loadData() .distinct() .map() .filter() .handle().subscribe();;//加载最新的商品浏览数据3、Reactor Core1、HttpHandler、HttpServer** * 测试webflux * @author : jucunqi * @since : 2025/1/16 */ public class FluxMainApplication { public static void main(String[] args) throws IOException { HttpHandler handler = (ServerHttpRequest request, ServerHttpResponse response) -> { URI uri = request.getURI(); System.out.println(Thread.currentThread() + "请求进来: " + uri); //编写请求处理的业务,给浏览器写一个内容 URL + "Hello~!" // response.getHeaders(); //获取响应头 // response.getCookies(); //获取Cookie // response.getStatusCode(); //获取响应状态码； // response.bufferFactory(); //buffer工厂 // response.writeWith() //把xxx写出去 // response.setComplete(); //响应结束 //创建 响应数据的 DataBuffer DataBufferFactory factory = response.bufferFactory(); String result = "Hello world"; //数据Buffer DataBuffer buffer = factory.wrap(result.getBytes(StandardCharsets.UTF_8)); // 需要一个 DataBuffer 的发布者 return response.writeWith(Flux.just(buffer)); }; //2、启动一个服务器，监听8080端口，接受数据，拿到数据交给 HttpHandler 进行请求处理 ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(handler); //3、启动Netty服务器 HttpServer.create() .host("localhost") .port(8080) .handle(adapter) //用指定的处理器处理请求 .bindNow(); //现在就绑定 System.out.println("服务器启动完成....监听8080，接受请求"); System.in.read(); System.out.println("服务器停止...."); } }4、DispatcherHandlerSpringMVC： DispatcherServlet；SpringWebFlux： DispatcherHandler1、请求处理流程HandlerMapping：请求映射处理器； 保存每个请求由哪个方法进行处理HandlerAdapter：处理器适配器；反射执行目标方法HandlerResultHandler：处理器结果处理器；SpringMVC： DispatcherServlet 有一个 doDispatch() 方法，来处理所有请求；WebFlux： DispatcherHandler 有一个 handle(ServerWebExchange exchange) 方法，来处理所有请求；public Mono<Void> handle(ServerWebExchange exchange) { if (this.handlerMappings == null) { return createNotFoundError(); } if (CorsUtils.isPreFlightRequest(exchange.getRequest())) { return handlePreFlight(exchange); } return Flux.fromIterable(this.handlerMappings) //拿到所有的 handlerMappings .concatMap(mapping -> mapping.getHandler(exchange)) //找每一个mapping看谁能处理请求 .next() //直接触发获取元素； 拿到流的第一个元素； 找到第一个能处理这个请求的handlerAdapter .switchIfEmpty(createNotFoundError()) //如果没拿到这个元素，则响应404错误； .onErrorResume(ex -> handleDispatchError(exchange, ex)) //异常处理，一旦前面发生异常，调用处理异常 .flatMap(handler -> handleRequestWith(exchange, handler)); //调用方法处理请求，得到响应结果 }1、请求和响应都封装在 ServerWebExchange 对象中，由handle方法进行处理2、如果没有任何的请求映射器； 直接返回一个： 创建一个未找到的错误； 404； 返回Mono.error；终结流3、跨域工具，是否跨域请求，跨域请求检查是否复杂跨域，需要预检请求；4、Flux流式操作，先找到HandlerMapping，再获取handlerAdapter，再用Adapter处理请求，期间的错误由onErrorResume触发回调进行处理；源码中的核心两个：handleRequestWith： 编写了handlerAdapter怎么处理请求handleResult： String、User、ServerSendEvent、Mono、Flux ...concatMap： 先挨个元素变，然后把变的结果按照之前元素的顺序拼接成一个完整流private <R> Mono<R> createNotFoundError() { Exception ex = new ResponseStatusException(HttpStatus.NOT_FOUND); return Mono.error(ex); } Mono.defer(() -> { Exception ex = new ResponseStatusException(HttpStatus.NOT_FOUND); return Mono.error(ex); }); //有订阅者，且流被激活后就动态调用这个方法； 延迟加载； 5、注解开发1、目标方法传参https://docs.spring.io/spring-framework/reference/6.0/web/webflux/controller/ann-methods/arguments.htmlController method argumentDescriptionServerWebExchange封装了请求和响应对象的对象; 自定义获取数据、自定义响应ServerHttpRequest, ServerHttpResponse请求、响应WebSession访问Session对象java.security.Principal org.springframework.http.HttpMethod请求方式java.util.Locale国际化java.util.TimeZone + java.time.ZoneId时区@PathVariable路径变量@MatrixVariable矩阵变量@RequestParam请求参数@RequestHeader请求头；@CookieValue获取Cookie@RequestBody获取请求体，Post、文件上传HttpEntity封装后的请求对象@RequestPart获取文件上传的数据 multipart/form-data.java.util.Map, org.springframework.ui.Model, and org.springframework.ui.ModelMap.Map、Model、ModelMap@ModelAttribute Errors, BindingResult数据校验，封装错误SessionStatus + class-level @SessionAttributes UriComponentsBuilderFor preparing a URL relative to the current request’s host, port, scheme, and context path. See URI Links.@SessionAttribute @RequestAttribute转发请求的请求域数据Any other argument所有对象都能作为参数：1、基本类型 ，等于标注@RequestParam 2、对象类型，等于标注 @ModelAttribute2、返回值写法sse和websocket区别：SSE：单工；请求过去以后，等待服务端源源不断的数据websocket：双工： 连接建立后，可以任何交互；Controller method return valueDescription@ResponseBody把响应数据写出去，如果是对象，可以自动转为jsonHttpEntity, ResponseEntityResponseEntity：支持快捷自定义响应内容HttpHeaders没有响应内容，只有响应头ErrorResponse快速构建错误响应ProblemDetailSpringBoot3；String就是和以前的使用规则一样；forward: 转发到一个地址redirect: 重定向到一个地址配合模板引擎View直接返回视图对象java.util.Map, org.springframework.ui.Model以前一样@ModelAttribute以前一样Rendering新版的页面跳转API； 不能标注 @ResponseBody 注解void仅代表响应完成信号Flux, Observable, or other reactive type使用 text/event-stream 完成SSE效果Other return values未在上述列表的其他返回值，都会当成给页面的数据；6、文件上传https://docs.spring.io/spring-framework/reference/6.0/web/webflux/controller/ann-methods/multipart-forms.htmlclass MyForm { private String name; private MultipartFile file; // ... } @Controller public class FileUploadController { @PostMapping("/form") public String handleFormUpload(MyForm form, BindingResult errors) { // ... } }现在@PostMapping("/") public String handle(@RequestPart("meta-data") Part metadata, @RequestPart("file-data") FilePart file) { // ... }7、错误处理 @ExceptionHandler(ArithmeticException.class) public String error(ArithmeticException exception){ System.out.println("发生了数学运算异常"+exception); //返回这些进行错误处理； // ProblemDetail： 建造者：声明式编程、链式调用 // ErrorResponse ： return "炸了，哈哈..."; }8、自定义Flux配置 WebFluxConfigurer容器中注入这个类型的组件，重写底层逻辑@Configuration public class MyWebConfiguration { //配置底层 @Bean public WebFluxConfigurer webFluxConfigurer(){ return new WebFluxConfigurer() { @Override public void addCorsMappings(CorsRegistry registry) { registry.addMapping("/**") .allowedHeaders("*") .allowedMethods("*") .allowedOrigins("localhost"); } }; } }9、Filter@Component public class MyWebFilter implements WebFilter { @Override public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) { ServerHttpRequest request = exchange.getRequest(); ServerHttpResponse response = exchange.getResponse(); System.out.println("请求处理放行到目标方法之前..."); Mono<Void> filter = chain.filter(exchange); //放行 //流一旦经过某个操作就会变成新流 Mono<Void> voidMono = filter.doOnError(err -> { System.out.println("目标方法异常以后..."); }) // 目标方法发生异常后做事 .doFinally(signalType -> { System.out.println("目标方法执行以后..."); });// 目标方法执行之后 //上面执行不花时间。 return voidMono; //看清楚返回的是谁！！！ } }3、R2DBC1、手写R2DBC用法：1、导入驱动: 导入连接池（r2dbc-pool）、导入驱动（r2dbc-mysql ）2、使用驱动提供的API操作引入依赖<dependency> <groupId>io.asyncer</groupId> <artifactId>r2dbc-mysql</artifactId> <version>1.0.5</version> </dependency>手写代码public static void main(String[] args) throws IOException { // 创建mysql配置 MySqlConnectionConfiguration configuration = MySqlConnectionConfiguration.builder() .host("localhost") .port(3306) .username("root") .password("12345678") .database("test") .build(); // 获取mysql连接工厂 MySqlConnectionFactory factory = MySqlConnectionFactory.from(configuration); Mono.from( factory.create() .flatMapMany(conn -> conn .createStatement("select * from customers where customer_id = ?") .bind(0, 1L) .execute() ).flatMap(result -> result.map(readable -> { return new Customers(((Integer) readable.get("customer_id")), Objects.requireNonNull(readable.get("customer_name")).toString()); })) ).subscribe(System.out::println); System.in.read(); }2、Spring Data R2DBC提升生产力方式的 响应式数据库操作0、整合1、导入依赖 <!-- https://mvnrepository.com/artifact/io.asyncer/r2dbc-mysql --> <dependency> <groupId>io.asyncer</groupId> <artifactId>r2dbc-mysql</artifactId> <version>1.0.5</version> </dependency> <!-- 响应式 Spring Data R2dbc--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-r2dbc</artifactId> </dependency>2、编写配置spring: r2dbc: password: 123456 username: root url: r2dbc:mysql://localhost:3306/test name: test3、@Autowired private R2dbcEntityTemplate template; /** * 测试template // 适合单表操作，复杂sql不好编写 * @throws IOException io异常 */ @Test public void springDataR2dbcTest() throws IOException { // 1. 构建查询条件 Criteria criteria = Criteria .empty() .and("project_leader") .is("1"); // 构建Query对象 Query query = Query .query(criteria); // 查询数据 template.select(query, com.jcq.r2dbc.eneity.Test.class) .subscribe(test -> System.out.println("test = " + test)); System.out.println(System.in.read()); } @Autowired private DatabaseClient databaseClient; /** * 测试databaseClient // 更底层，适合复杂sql 比如join */ @Test public void databaseClientTest() throws IOException { databaseClient.sql("select * from test where id in (?,?)") .bind(0, 1) .bind(1, 2) .fetch() // 抓取数据 .all() // 抓取所有数据 .map(a -> new com.jcq.r2dbc.eneity.Test(((Integer) a.get("id")),a.get("project_leader").toString())) .subscribe(a -> System.out.println("a = " + a)); System.out.println(System.in.read()); }1、声明式接口：R2dbcRepositoryRepository接口@Repository public interface TAutherRepository extends R2dbcRepository<TAuther,Long> { // 根据命名实现sql Flux<TAuther> findAllByIdAndNameLike(Long id,String name); @Query("select * from t_author") Flux<TAuther> queryList(); } 自定义Converter@ReadingConverter // 读取数据库的时候，吧row转成 TBook public class TBookConverter implements Converter<Row, TBook> { @Override public TBook convert(Row source) { TBook tBook = new TBook(); tBook.setId((Long) source.get("id")); tBook.setTitle((String) source.get("title")); tBook.setAuthorId((Long) source.get("author_id")); Object instance = source.get("publish_time"); System.out.println(instance); ZonedDateTime instance1 = (ZonedDateTime) instance; tBook.setPublishTime(instance1.toInstant()); TAuther tAuther = new TAuther(); tAuther.setName(source.get("name", String.class)); tBook.setTAuther(tAuther); return tBook; } }配置生效@Configuration public class R2DbcConfiguration { @Bean @ConditionalOnMissingBean public R2dbcCustomConversions r2dbcCustomConversions() { return R2dbcCustomConversions.of(MySqlDialect.INSTANCE, new TBookConverter()); } } 3、编程式组件R2dbcEntityTemplateDatabaseClient4、最佳实践最佳实践： 提升生产效率的做法1、Spring Data R2DBC，基础的CRUD用 R2dbcRepository 提供好了2、自定义复杂的SQL（单表）： @Query；3、多表查询复杂结果集： DatabaseClient 自定义SQL及结果封装；@Query + 自定义 Converter 实现结果封装经验：1-1:1-N 关联关系的封装都需要自定义结果集的方式Spring Data R2DBC：自定义Converter指定结果封装DatabaseClient：贴近底层的操作进行封装; 见下面代码MyBatis： 自定义 ResultMap 标签去来封装databaseClient.sql("select b.*,t.name as name from t_book b " + "LEFT JOIN t_author t on b.author_id = t.id " + "WHERE b.id = ?") .bind(0, 1L) .fetch() .all() .map(row-> { String id = row.get("id").toString(); String title = row.get("title").toString(); String author_id = row.get("author_id").toString(); String name = row.get("name").toString(); TBook tBook = new TBook(); tBook.setId(Long.parseLong(id)); tBook.setTitle(title); TAuthor tAuthor = new TAuthor(); tAuthor.setName(name); tAuthor.setId(Long.parseLong(author_id)); tBook.setAuthor(tAuthor); return tBook; }) .subscribe(tBook -> System.out.println("tBook = " + tBook)); -
java使用POI导出Excel单元格为数字类型 第一版开发的时候，所有的单元格都是文本类型，由于需求方需要导出Excel可以直接使用函数计算，所以需要改动一下，将导出的Excel设置为数值类型。创建XSSFCellStyle// 此处设置数据格式 XSSFDataFormat df = wb.createDataFormat(); // 创建单元格样式 XSSFCellStyle numberStyle = wb.createCellStyle(); numberStyle.setFillForegroundColor((short) 1); //设置要添加表背景颜色 numberStyle.setFillPattern(FillPatternType.SOLID_FOREGROUND); //solid 填充 numberStyle.setAlignment(XSSFCellStyle.ALIGN_CENTER); //文字水平居中 numberStyle.setVerticalAlignment(XSSFCellStyle.VERTICAL_CENTER);//文字垂直居中 numberStyle.setBorderBottom(BorderStyle.THIN); //底边框加黑 numberStyle.setBorderLeft(BorderStyle.THIN); //左边框加黑 numberStyle.setBorderRight(BorderStyle.THIN); // 有边框加黑 numberStyle.setBorderTop(BorderStyle.THIN); //上边框加黑 // 注意#,##0.00_ 后面有一个空格 numberStyle.setDataFormat(df.getFormat("#,##0.00_ ")); //为单元格添加背景样式 sheet=setTitleStyle(sheet,rowNum,colNum,style1);创建单元格，并且设置XSSFCellStyle。private static XSSFSheet setTitleStyle(XSSFSheet sheet,int rowNum,int colNum,XSSFCellStyle style){ for (int i = 0; i < rowNum; i++) { //需要行表格 Row row = sheet.createRow(i); //创建行 row.setHeight((short)600); for (int j = 0; j < colNum; j++) {//需要列 row.createCell(j).setCellStyle(style); } } return sheet; }在为单元格赋值的时候，将数据转换为Double类型。
-
使用Apache POI 设置单元格中文字方向 前几天遇到了一个需求，需要使用Apache POI导出Excel，并且还需要实现单元格合并和文字竖向展示的功能。最终结果是这个样子介绍一下合并单元格和文字竖向展示的实现方法。1、合并单元格只要知道需要合并单元格的行号和列号就可以//创建工作簿 XSSFWorkbook workBook = new XSSFWorkbook(); //创建一个sheet XSSFSheet sheet = workBook.createSheet(); CellRangeAddress balanceCell = new CellRangeAddress(0, 4, 1, 1); sheet.addMergedRegion(balanceCell);2、文字竖向显示获取到需要竖向显示的单元格，然后设置单元格样式，设置Rotation属性 //创建工作簿 XSSFWorkbook workBook = new XSSFWorkbook(); //创建一个sheet XSSFSheet sheet = workBook.createSheet(); CellStyle directionStyle = workBook.createCellStyle(); directionStyle.setRotation((short)255); XSSFRow row = sheet.getRow(0); XSSFCell cell = row.getCell(4); cell.setCellStyle(directionStyle);
-
MyBatis分页助手报错java.util.ArrayList cannot be cast to com.github.pagehelper.Page 最近在使用MyBatis时遇到了这问题原本可以正常使用分页，但是当我添加了一个查询条件时突然报出了这个错误，后来终于找到了原因。因为PageHelper.startPage(pageNum,pageSize)方法 只会对靠近的第一个查询语句进行分页。结果我新添加的查询条件又进行了一个select查询，所以会产生这个错误如：PageHelper.startPage(pageNum, pageSize, true);XXXMapper.xxxxPage<> page= (Page<>)TestMapper.query( );改为：XXXMapper.xxxxPageHelper.startPage(pageNum, pageSize, true);Page<> page= (Page<>)TestMapper.query( ); 或者：PageHelper.startPage(pageNum, pageSize, true);Page<> page= (Page<>)TestMapper.query( );XXXMapper.xxxx这样就不会报这个错了。 还是因为自己对这个分页助手不够了解呀。。。
-
写给新同学的SpringBoot教程 — 高级篇 写给新同学的SpringBoot教程 — 高级篇一、添加缓存1、使用SpringBoot自带缓存功能1、基本使用在主类上添加注解@EnableCaching在业务方法上添加@Cacheable(cacheNames = "xxx")注解这样即可实现基础的缓存功能基本流程：@Cacheable:方法执行之前，先去查询Cache（缓存组件），按照CacheNames指定的名字获取，（CacheManager先获取相应的缓存），第一次获取缓存如果没有Cache组件会自行创建去Cache中查找缓存的内容，使用一个key，默认就是方法的参数key是按照某种策略生成的，默认hi使用keyGenerator生成的。默认使用SimplekeyGenerator生成key： 如果有一个参数，key=参数的值 如果没有参数，key=new SimpleKey(); 如果有多个参数，key=new SimpleKey(params);没有查到缓存就去调用目标方法；方法查询结束后，将目标方法返回的结果放入到缓存中。总结： @Cacheable标注的方法执行之前先来检查缓存中有没有这个数据，默认按照参数的值作为key去查询缓存，如果缓存中没有，就执行方法并将结果放入到缓存中；以后再来调用就可以直接将缓存中的数据进行返回。核心：使用CaCheManager【ConCurrentMapCacheManager】按照名字得到Cache【ConcurrentMapCache】组件key是使用keyGenerator生成的，默认是SimpleKeyGenerator主类：@SpringBootApplication @MapperScan(value = "com.jcq.cache.mapper") @EnableCaching //这个注解不要忘了 public class Springboot09CacheApplication { public static void main(String[] args) { SpringApplication.run(Springboot09CacheApplication.class, args); } }业务方法@Service public class EmployeeService { @Resource private EmployeeMapper employeeMapper; @Cacheable(cacheNames = "emp",key = "#root.methodName+'['+#id+']'") public Employee getEmp(Integer id){ return employeeMapper.selEmpByid(id); } }2、其他属性keyGenerator：key的生成器，可以自己指定key的生成器组件的id ，key/keyGenerator二选一使用。cacheManager:指定缓存管理器，或者cacheResolver指定获取解析器。condition：满足条件是放入缓存。unless：满足条件是不放入缓存，与condition正好相反。sync：默认为false，true为开启异步缓存。若开启异步，unless注解无效3、其他注解@CachePut注解：方法运行完后，将数据放入缓存中@CachePut与@Cacheable的区别： 后者在方法运行前先到缓存总查询，前者在方法执行后，将数据放入缓存中。注意： 只有保持数据的key相同，才能实现数据的同步更新。 @CachePut(cacheNames = "emp",key = "#employee.id") public Employee update(Employee employee){ employeeMapper.updEmp(employee); return employee; }@CacheEvict:清除缓存数据allEntries:清空所有缓存，默认为falsebeforeInvocation：在方法执行前清空缓存，默认为false注意：保持数据key同一，才可以保持数据同步。 //测试清除缓存中的数据 @CacheEvict(cacheNames = "emp",key = "#id") public void delEmpCache(Integer id){ System.out.println("清除缓存"); }@Caching：可以将多个注解组合//测试Caching的使用 @Caching(cacheable = { @Cacheable(value = "emp", key = "#lastName") }, put = {@CachePut(value = "emp", key = "#result.id"), @CachePut(value = "emp",key = "#result.email") } ) public Employee selBylastName(String lastName){ return employeeMapper.selBylastName(lastName); } 2、整合redis1、redis命令常用命令网址2、步骤在pom.xml中引入下列依赖<!--引入redis启动器--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency>配置文件如下#配置redis spring.redis.host=192.168.91.128SpringBoot把对redis的操作封装在RedisTemplate 和 StringRedisTemplate中@SpringBootTest class Springboot09CacheApplicationTests { @Autowired private RedisTemplate redisTemplate; @Autowired private StringRedisTemplate stringRedisTemplate; @Autowired private EmployeeService employeeService; @Test public void testRedis(){ //添加字符串 // stringRedisTemplate.opsForValue().append("msg","helloworld"); //添加list // stringRedisTemplate.opsForList().leftPushAll("mylist","1","2","3"); //取出list // String str = stringRedisTemplate.opsForList().leftPop("mylist"); // System.out.println(str); Employee emp = employeeService.getEmp(1); //添加对象 redisTemplate.opsForValue().set("emp",emp); }3、添加自定义组件，将对象转化为json字符串存储@Configuration public class MyRedisTemplate { @Bean public RedisTemplate<Object, Employee> redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException { RedisTemplate<Object, Employee> template = new RedisTemplate(); template.setConnectionFactory(redisConnectionFactory); template.setDefaultSerializer(new Jackson2JsonRedisSerializer(Employee.class)); return template; } }而在服务启动过程中，使用的CacheManager依旧默认按照序列化的方式进行对象存储，所以还需要添加组件来达到将对象用json存储的目的 /** * 缓存管理器 */ @Bean public CacheManager cacheManager(RedisConnectionFactory redisConnectionFactory) { //初始化一个RedisCacheWriter RedisCacheWriter redisCacheWriter = RedisCacheWriter.nonLockingRedisCacheWriter(redisConnectionFactory); //设置CacheManager的值序列化方式为json序列化 RedisSerializer<Object> jsonSerializer = new GenericJackson2JsonRedisSerializer(); RedisSerializationContext.SerializationPair<Object> pair = RedisSerializationContext.SerializationPair .fromSerializer(jsonSerializer); RedisCacheConfiguration defaultCacheConfig=RedisCacheConfiguration.defaultCacheConfig() .serializeValuesWith(pair); //设置默认超过期时间是30秒 defaultCacheConfig.entryTtl(Duration.ofSeconds(30)); //初始化RedisCacheManager return new RedisCacheManager(redisCacheWriter, defaultCacheConfig); }二、SpringBoot与消息1、使用docker安装rabbitmq//下载rabbitmq docker pull rabbitmq:3-management //运行 docker run -d -p 5672:5672 -p 15672:15672 --name myrabbitmq id默认用户名密码均为：guest转换器类型：direct:只有key相同的队列才可以收到消息fanout:任意队列都能收到消息atguigu.# : #表示0或多个词.news : 表示0或一个单词2、SpringBoot整合Rabbitmq1、基础使用配置文件：**spring.rabbitmq.username=guest spring.rabbitmq.password=guest spring.rabbitmq.host=192.168.91.128 #默认为5672，不需要配置 #spring.rabbitmq.port=5672测试代码： /** * 单播 */ @Test void contextLoads() { Map<String, Object> map = new HashMap<>(); map.put("name", "张三"); map.put("age", 18); rabbitTemplate.convertAndSend("exchange.direct","atguigu.news",map); }但是这样发送的消息，会以默认的序列化方式进行存储。效果接收消息： /** * 接收消息 */ @Test void recieve(){ Object o = rabbitTemplate.receiveAndConvert("atguigu.news"); System.out.println(o.getClass()); System.out.println(o); }自定义组件，使用json格式进行对象转化@Configuration public class MyAMQPConfig { @Bean public MessageConverter messageConverter(){ return new Jackson2JsonMessageConverter(); } }广播测试代码 /** * 广播 */ @Test void sendMsg(){ rabbitTemplate.convertAndSend("exchange.fanout","",new Book("SpringMVC","张佳明")); }2、消息监听主类上添加@EnableRabbit注解@EnableRabbit @SpringBootApplication public class SpringBoot10AmqpApplication { public static void main(String[] args) { SpringApplication.run(SpringBoot10AmqpApplication.class, args); } }业务代码如下：@Service public class BookService { @RabbitListener(queues = "atguigu.news") public void bookLinstener(Book book){ System.out.println("接收到消息"+book); } }3、使用AmqpAdmin管理工具 @Autowired private AmqpAdmin amqpAdmin; @Test void createRabbit(){ //创建转换器 // amqpAdmin.declareExchange(new DirectExchange("AmqpAdmin.exchange")); // System.out.println("转换器创建成功"); //创建消息队列 // amqpAdmin.declareQueue(new Queue("AmqpAdmin.news")); // System.out.println("消息队列创建成功"); //创建绑定规则 amqpAdmin.declareBinding(new Binding("AmqpAdmin.news", Binding.DestinationType.QUEUE, "AmqpAdmin.exchange", "AmqpAdmin.haha", null)); System.out.println("创建绑定成功"); //删除队列 amqpAdmin.deleteQueue("AmqpAdmin.news"); System.out.println("删除队列"); amqpAdmin.deleteExchange("AmqpAdmin.exchange"); System.out.println("删除转换器"); }三、SpringBoot与检索ElasticSearch1、搭建环境1、下载es docker pull elasticsearch 2、启动命令 [root@localhost ~]# docker run -e ES_JAVA_OPTS="-Xms256m -Xmx256m" -d -p 9200:9200 -p 9300:9300 --name ES01 5acf0e8da90b 7d33661ebabc32621173c4c9648047016e6ede9fc625c51340692736b7fdf170 其中: ES_JAVA_OPTS="-Xms256m -Xmx256m" :-Xms 初始堆内存大小 -Xmx 最大使用内存 9200：es默认web通信使用9200端口 9300：当分布式情况下es多个节点之间通信使用9300端口 文档地址：https://www.elastic.co/guide/cn/elasticsearch/guide/current/index-doc.htmlSpringBoot默认支持两种技术和ElasticSearch交互一、Jest（默认不生效）需要导入jest的工具包（io.serachbox.chlient.JestClient）需要将默认的SpringBoot-data-Elasticsearch注释掉导入jest依赖但是jest已经过时，不演示了二、SpringDataClient节点信息clusterNodes,clusterNameElasticsearchTemplate来操作es编写一个ElasticsearchRepository子接口来操作es因版本问题，暂时没有解决SpringBoot2.X版本整合es的方法，使用会报错四、SpringBoot与任务1、异步任务1、在主类添加@EnableAsync注解2、在业务层添加@Async注解即可实现异步任务@Async @Service public class AsyncService { public void testAsync(){ try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("方法执行完毕"); } }2、定时任务cron表达式：Cron表达式的格式：秒 分 时 日 月 周 年(可选)。字段名 允许的值 允许的特殊字符 秒 0-59 , - * / 分 0-59 , - * / 小时 0-23 , - * / 日 1-31 , - * ? / L W C月 1-12 or JAN-DEC , - * / 周几 1-7 or SUN-SAT , - * ? / L C # 年(可选字段) empty 1970-2099 , - * /具体的用法：https://www.cnblogs.com/lazyInsects/p/8075487.html @Scheduled(cron = "0 * * * * MON-FRI") 周一到周五，没整分运行一次 // @Scheduled(cron = "0/4 * * * * MON-FRI") 没整分，间隔4秒执行 @Scheduled(cron = ) public void testAsync(){ System.out.println("方法执行完毕"); } }在方法上加上@Scheduled注解在主类添加@EnableSchelding注解主类代码如下//@EnableAsync @EnableScheduling @SpringBootApplication public class SpringBoot12TaskApplication { public static void main(String[] args) { SpringApplication.run(SpringBoot12TaskApplication.class, args); } } 3、邮件任务配置文件spring.mail.username=1036658425@qq.com spring.mail.password=kersbnhzzuhfbdih spring.mail.host=smtp.qq.com@Autowired JavaMailSenderImpl mailSender; @Test void contextLoads() { //创建简单邮件对象 SimpleMailMessage massage = new SimpleMailMessage(); //设置主题 massage.setSubject("简单测试"); //设置内容 massage.setText("测试简单邮件发送"); //设置收件方 massage.setTo("594082079@qq.com"); //设置发送方 massage.setFrom("1036658425@qq.com"); mailSender.send(massage); } @Test public void test02() throws MessagingException { //测试复杂邮件发送 MimeMessage mimeMessage = mailSender.createMimeMessage(); MimeMessageHelper helper = new MimeMessageHelper(mimeMessage,true); helper.setSubject("复杂发送"); helper.setText("<b style='color:red'>测试复杂邮件发送</b>",true); helper.addAttachment("a.jpg",new File("E:\\笔记\\SpringBoot\\springboot高级\\SpringBoot高级.assets\\image-20200310195943079.png")); helper.addAttachment("b.jpg",new File("E:\\笔记\\SpringBoot\\springboot高级\\SpringBoot高级.assets\\image-20200310201323308.png")); helper.setTo("594082079@qq.com"); helper.setFrom("1036658425@qq.com"); mailSender.send(mimeMessage); } 五、SpringBoot与安全1、业务部分代码编写配置类@EnableWebSecurity public class MySecurityConfig extends WebSecurityConfigurerAdapter { //添加认证功能 @Override protected void configure(HttpSecurity http) throws Exception { // super.configure(http); http.authorizeRequests().mvcMatchers("/").permitAll() .mvcMatchers("/level1/**").hasRole("VIP1") .mvcMatchers("/level2/**").hasRole("VIP2") .mvcMatchers("/level3/**").hasRole("VIP3"); //开启登录认证页面 http.formLogin(); //开启自动配置注销功能 http.logout().logoutSuccessUrl("/"); //访问/logout 清除session //开启自动配置记住我功能 http.rememberMe(); } //添加登录授权功能 @Override protected void configure(AuthenticationManagerBuilder auth) throws Exception { // super.configure(auth); auth.inMemoryAuthentication().passwordEncoder(new BCryptPasswordEncoder()) .withUser("zhangsan").password(new BCryptPasswordEncoder().encode("123456")).roles("VIP1","VIP2") .and().withUser("lisi").password(new BCryptPasswordEncoder().encode("123456")).roles("VIP2","VIP3") .and().withUser("wangwu").password(new BCryptPasswordEncoder().encode("123456")).roles("VIP3","VIP1"); }整合页面部分，因版本问题，暂时没有效果，先不放图片了六、分布式1、Dubbo下载镜像[root@localhost ~]# docker pull zookeeper [root@localhost ~]# docker run --name zk01 -p 2181:2181 --restart always -d bbebb888169c 1、提供者将服务提供者注册到注册中心导入相关依赖Dubbo和Zkclient<!--引入Dubbo服务启动器--> <dependency> <groupId>com.alibaba.boot</groupId> <artifactId>dubbo-spring-boot-starter</artifactId> <version>0.1.0</version> </dependency> <!--引入zookeeper的客户端工具--> <!-- https://mvnrepository.com/artifact/com.github.sgroschupf/zkclient --> <dependency> <groupId>com.github.sgroschupf</groupId> <artifactId>zkclient</artifactId> <version>0.1</version> </dependency>2. 在主类添加启用Dubbo的注解 @EnableDubbo @SpringBootApplication public class ProviderTicketApplication { public static void main(String[] args) { SpringApplication.run(ProviderTicketApplication.class, args); } }如果不加此注解，消费者会报空指针异常 2. 配置application配置文件 dubbo.application.name=provider dubbo.registry.address=zookeeper://192.168.91.128:2181 dubbo.scan.base-packages=com.jcq.ticket.service使用@service发布服务@Component @Service public class TickerServiceImpl implements TickerService { @Override public String getTicker() { return "《厉害了，我的国》"; } }2、消费者导入相关依赖<!--引入Dubbo服务启动器--> <dependency> <groupId>com.alibaba.boot</groupId> <artifactId>dubbo-spring-boot-starter</artifactId> <version>0.1.0</version> </dependency> <!--引入zookeeper的客户端工具--> <!-- https://mvnrepository.com/artifact/com.github.sgroschupf/zkclient --> <dependency> <groupId>com.github.sgroschupf</groupId> <artifactId>zkclient</artifactId> <version>0.1</version> </dependency>配置文件dubbo.application.name=consumer dubbo.registry.address=zookeeper://192.168.91.128:2181 server.port=8081引用服务需要把提供者的包个接口复制到项目中，且保持路径同一加下来就是编写业务代码了import com.alibaba.dubbo.config.annotation.Reference; import org.springframework.stereotype.Service; @Service public class UserServiceImpl { @Reference private TickerService tickerService; public String getTicker(){ return tickerService.getTicker(); } }注意不要到错包了，会导致空指针异常。2、SpringCloud1、注册中心1、首先在主类添加@EnableEurekaServer注解@EnableEurekaServer @SpringBootApplication public class EurekaServerApplication { public static void main(String[] args) { SpringApplication.run(EurekaServerApplication.class, args); } }2、添加配置文件server: port: 8761 #修改服务端口 eureka: client: fetch-registry: false #不从注册中心获取服务 register-with-eureka: false #不在服务中心注册服务 service-url: defaultZone: http://localhost:8761/eureka/ #访问路径 instance: hostname: eureka-server #主机名称3、登录页面查看是否启动成功网址：http://localhost:8761/2、提供者1、编写配置文件server: port: 8001 spring: application: name: provider-ticket eureka: instance: prefer-ip-address: true #注册服务时，使用服务的ip地址 service-url: defaultZone: http://localhost:8761/eureka/2、正常编写业务代码及控制器代码一个项目注册多个服务直接将不同端口好的项目进行打包，使用cmd命令分别执行即可项目打包后的图片管理页面显示内容入下3、消费者1、编写配置文件server: port: 8200 eureka: instance: prefer-ip-address: true service-url: defaultZone: http://localhost:8761/eureka/ spring: application: name: consumer-user2、主类代码：@EnableDiscoveryClient //开启到注册中心查找服务 @SpringBootApplication public class ConsumerUserApplication { public static void main(String[] args) { SpringApplication.run(ConsumerUserApplication.class, args); } /** * 给容器添加组件 * 可以发送远程http请求 * @return */ @LoadBalanced //开启负载均衡功能 @Bean public RestTemplate restTemplate(){ return new RestTemplate(); } }3、业务代码@RestController public class UserController { @Autowired private RestTemplate restTemplate; @GetMapping("/buy") public String buyTicket(String name){ //发送远程请求，请求注册中心的业务方法。地址直接写注册中心中的服务名，不需写端口号 String ticket = restTemplate.getForObject("http://PROVIDER-TICKET/get", String.class); return name+"购买了"+ticket; } }七、热部署只需要到如SpringBoot提供的依赖即可<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-devtools</artifactId> <scope>runtime</scope> <optional>true</optional> </dependency>更改代码后，手动CTRL+F9 重新编译，即可实现热部署
