总阅读量:次

⭐⭐⭐ Spring Boot 项目实战 ⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》 《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》 《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》 《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》 《Java 面试题 + Java 学习指南》

摘要: 原创出处 blog.csdn.net/m0_64360721/article/details/122960311 「代码峡谷孙膑」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

  1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
  2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
  3. 您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。
  4. 新的源码解析文章实时收到通知。每周更新一篇左右。
  5. 认真的源码交流微信群。

介绍

下载功能应该是比较常见的功能了，虽然一个项目里面可能出现的不多，但是基本上每个项目都会有，而且有些下载功能其实还是比较繁杂的，倒不是难，而是麻烦。

所以结合之前的下载需求，我写了一个库来简化下载功能的实现

❝

传送门：https://github.com/Linyuzai/concept/wiki/Concept-Download

❞

如果我说现在只需要一个注解就能帮你下载任意的对象，是不是觉得非常的方便

@Download(source = "classpath:/download/README.txt")
@GetMapping("/classpath")
public void classpath() {

}

@Download
@GetMapping("/file")
public File file() {
    return new File("/Users/Shared/README.txt");
}

@Download
@GetMapping("/http")
public String http() {
    return "http://127.0.0.1:8080/concept-download/image.jpg";
}

感觉差别不大？那就听听我遇到的一个下载需求

我们有一个平台是管理设备的，然后每个设备都会有一个二维码图片，用一个字段存储的 http 地址

现在需要导出所有设备二维码图片的压缩包，图片名称需要用设备名称加 .png 后缀，需求上来说并不难，但是着实有点麻烦

  • 首先我需要将设备列表查出来
  • 然后使用二维码地址下载图片并写到本地缓存文件
  • 在下载之前需要先判断是否已经存在缓存
  • 下载时需要并发下载提升性能
  • 等所有图片下载结束后
  • 再生成一个压缩文件
  • 然后再操作输入输出流写到响应中

看着我实现了将近 200 行的代码，真是又臭又长，一个下载功能咋能那么麻烦呢，于是我就想有没有更简单的方式

我当时的需求很简单，我想着我只要提供需要下载的数据，比如一个文件路径，一个文件对象，一段字符串文本，一个http地址，或者混搭了前面所有类型的一个集合，甚至是我们自定义的某个类的实例，后面的事情我就不用管了

文件路径是一个文件还是一个目录？字符串文本需要先写入一个文本文件中？http资源如何下载到本地？多个文件怎么压缩？最后怎么写到响应中？我才不想花时间管这些

比如就像我现在这个需求，我只要返回设备列表就行了，其他的事情我都不用管

@Download(filename = "二维码.zip")
@GetMapping("/download")
public List<Device> download() {
    return deviceService.all();
}

public class Device {

    //设备名称
    private String name;

    //设备二维码
    //注解表示该http地址是需要下载的数据
    @SourceObject
    private String qrCodeUrl;

    //注解表示文件名称
    @SourceName
    public String getQrCodeName() {
        return name + ".png";
    }
    //省略其他属性方法
}

通过在 Device 的字段上标注某些注解（或是实现某个接口）来指定文件名称和文件地址

如果能这样实现，省时省心省力，又多了写 199 行代码的摸鱼时间难道不香么

思路

下面来讲讲这个库的主要设计思路，以及中间遇到的坑，大家有兴趣可以继续往下看

其实基于一开始的设想，我觉得功能并没有多复杂，于是就决定开肝

只是万万没想到实现起来比我想象的更复杂（这是后话了）

基础

首先整个库基于响应式编程，但却并不是完全意义上的响应式，只能说是Mono<InputStream>这样的。。。奇怪组合？

为什么会这样呢，很大的一个原因是由于需要兼容webmvc和webflux，导致我仅仅是将之前实现的InputStream方式重构成了响应式，所以就出现了这样的组合

这也是我遇到的最大的一个坑，我先前已经基本调通了基于Servlet的整个下载流程，然后就想着支持一下webflux

大家都知道webmvc中，我们可以通过RequestContextHolder来获得请求和响应对象，但是在webflux中就不行了，当然我们可以在方法参数中注入

@Download(source = "classpath:/download/README.txt")
@GetMapping("/classpath")
public void classpath(ServerHttpResponse response) {

}

结合Spring自带的注入功能，我们就可以通过AOP拿到响应的入参了，但是总觉得这样写有点多余，强迫症表示不能忍

有什么办法既能把用不到的入参干掉，又能拿到响应对象呢，在网上找到了一种实现方式

/**
 * 用于设置当前的请求和响应。
 *
 * @see ReactiveDownloadHolder
 */
public class ReactiveDownloadFilter implements WebFilter {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        ServerHttpResponse response = exchange.getResponse();
        return chain.filter(exchange)
                //低版本使用subscriberContext
                .contextWrite(ctx -> ctx.put(ServerHttpRequest.class, request))
                .contextWrite(ctx -> ctx.put(ServerHttpResponse.class, response));
    }
}

/**
 * 用于获得当前的请求和响应。
 *
 * @see ReactiveDownloadFilter
 */

public class ReactiveDownloadHolder {

    public static Mono<ServerHttpRequest> getRequest() {
        //低版本使用subscriberContext
        return Mono.deferContextual(contextView -> Mono.just(contextView.get(ServerHttpRequest.class)));
    }

    public static Mono<ServerHttpResponse> getResponse() {
        //低版本使用subscriberContext
        return Mono.deferContextual(contextView -> Mono.just(contextView.get(ServerHttpResponse.class)));
    }
}

通过添加WebFilter就可以获得响应对象了，但是返回值是Mono<ServerHttpResponse>

那么可不可以通过Mono.block()阻塞得到对应的对象呢，答案是不行，由于webflux基于Netty的非阻塞线程，如果调用该方法会直接抛出异常

所以就没有任何办法了，只能将之前代码基于响应式重构

架构

接下来说说整体架构

对于一个下载请求，我们可以分成几个步骤，以下载多个文件的压缩包为例

  • 首先我们一般是得到多个文件的路径或对应的File对象
  • 然后将这些文件压缩生成一个压缩文件
  • 最后将压缩文件写入到响应中

但是对于我上面描述的需求，一开始就不是文件路径或对象了，而是一个http地址，然后在压缩之前还需要多一个步骤，需要先将图片下载下来

那么对于各种各样的需求我们可能需要在当前步骤中的任意位置添加额外的步骤，所以我参考了Spring Cloud Gateway 拦截链的实现方式

/**
 * 下载处理器。
 */
public interface DownloadHandler extends OrderProvider {

  /**
   * 执行处理。
   *
   * @param context {@link DownloadContext}
   * @param chain   {@link DownloadHandlerChain}
   */
  Mono<Void> handle(DownloadContext context, DownloadHandlerChain chain);
}

/**
 * 下载处理链。
 */
public interface DownloadHandlerChain {

  /**
   * 调度下一个下载处理器。
   *
   * @param context {@link DownloadContext}
   */
  Mono<Void> next(DownloadContext context);
}

这样每个步骤就可以单独实现一个DownloadHandler，步骤与步骤之间可以任意的组合添加

下载上下文

在此基础上使用一个贯穿整个流程的上下文DownloadContext，方便共享和传递步骤之间的中间结果

对于上下文DownloadContext也提供了DownloadContextFactory可以用于自定义上下文

同时提供了DownloadContextInitializer和DownloadContextDestroyer用于在上下文初始化和销毁时扩展自己的逻辑

下载类型支持

我们需要下载的数据的类型是不固定的，比如有文件，有http地址，也会有之前我希望的自定义的类的实例

所以我将所有的下载对象抽象成了Source，表示一个下载源，这样文件可以实现为FileSource，http地址可以实现为HttpSource，然后通过对应的SourceFactory来匹配创建

比如FileSourceFactory可以匹配File并且创建FileSource，HttpSourceFactory可以匹配http://前缀并且创建HttpSource

/**
 * {@link Source} 工厂。
 */
public interface SourceFactory extends OrderProvider {

  /**
   * 是否支持需要下载的原始数据对象。
   *
   * @param source  需要下载的原始数据对象
   * @param context {@link DownloadContext}
   * @return 如果支持则返回 true
   */
  boolean support(Object source, DownloadContext context);

  /**
   * 创建。
   *
   * @param source  需要下载的原始数据对象
   * @param context {@link DownloadContext}
   * @return 创建的 {@link Source}
   */
  Source create(Object source, DownloadContext context);
}

那么对于我们自定义的类要怎么支持呢，之前提到可以在类上标注注解或是实现特定的接口，那么就用我实现的注解的方式来大概讲一讲吧

其实逻辑很简单，只要能熟练的运用反射就完全没问题，我们再来看一看用法

@Download(filename = "二维码.zip")
@GetMapping("/download")
public List<Device> download() {
        return deviceService.all();
        }

public class Device {

  //设备名称
  private String name;

  //设备二维码
  //注解表示该http地址是需要下载的数据
  @SourceObject
  private String qrCodeUrl;

  //注解表示文件名称
  @SourceName
  public String getQrCodeName() {
    return name + ".png";
  }
  //省略其他属性方法
}

首先我定义了一个注解@SourceModel标注在类上表示需要被解析，然后定义了一个@SourceObject注解标注在需要下载的字段（或方法）上，这样我们就可以通过反射拿到这个字段（或方法）的值

基于当前支持的SourceFactory就能创建出对应的Source，接下来使用@SourceName指定名称，也同样可以通过反射获得这个方法（或字段）的值并依旧通过反射设置到创建出来的Source上

这样就能非常灵活的支持任意的对象类型了

并发加载

对于像http这种网络资源，我们需要先并发加载（多个文件时）到本地的内存中或是缓存文件中来提升我们的处理效率

当然我可以直接定死一个线程池来执行，但是每个机器每个项目甚至每个需求对于并发的要求和资源的分配都不一样

所以我提供了SourceLoader来支持自定义的加载逻辑，你甚至可以一部分用线程池，一部分用协程，剩下一部分不加载

/**
 * {@link Source} 加载器。
 *
 * @see DefaultSourceLoader
 * @see SchedulerSourceLoader
 */
public interface SourceLoader {

  /**
   * 执行加载。
   *
   * @param source  {@link Source}
   * @param context {@link DownloadContext}
   * @return 加载后的 {@link Source}
   */
  Mono<Source> load(Source source, DownloadContext context);
}

压缩

当我们加载完之后就可以执行压缩了，同样的我定义了一个类Compression作为压缩对象的抽象

一般来说，我们会先在本地创建一个缓存文件，然后将压缩后的数据写入到缓存文件中

不过我每次都很讨厌在配置文件中配置各种各样的路径，所以在压缩时支持内存压缩，当然如果文件比较大还是老老实实生成一个缓存文件

对于压缩格式也提供了可以完全自定义的SourceCompressor接口，你想自己实现一个压缩协议都没有问题

/**
 * {@link Source} 压缩器。
 *
 * @see ZipSourceCompressor
 */
public interface SourceCompressor extends OrderProvider {

  /**
   * 获得压缩格式。
   *
   * @return 压缩格式
   */
  String getFormat();

  /**
   * 判断是否支持对应的压缩格式。
   *
   * @param format  压缩格式
   * @param context {@link DownloadContext}
   * @return 如果支持则返回 true
   */
  default boolean support(String format, DownloadContext context) {
    return format.equalsIgnoreCase(getFormat());
  }

  /**
   * 如果支持对应的格式就会调用该方法执行压缩。
   *
   * @param source  {@link Source}
   * @param writer  {@link DownloadWriter}
   * @param context {@link DownloadContext}
   * @return {@link Compression}
   */
  Compression compress(Source source, DownloadWriter writer, DownloadContext context);
}

响应写入

我将响应抽象成了DownloadResponse，主要用于兼容HttpServletResponse和ServerHttpResponse

但是问题又出现了，下面是webmvc和webflux写入响应的方式

//HttpServletResponse
response.getOutputStream().write(byte b[], int off, int len);

//ServerHttpResponse
        response.writeWith(Publisher<? extends DataBuffer> body);

这兼容的我脑壳疼，不过最后还是搞定了

/**
 * 持有 {@link ServerHttpResponse} 的 {@link DownloadResponse}，用于 webflux。
 */
@Getter
public class ReactiveDownloadResponse implements DownloadResponse {

  private final ServerHttpResponse response;

  private OutputStream os;

  private Mono<Void> mono;

  public ReactiveDownloadResponse(ServerHttpResponse response) {
    this.response = response;
  }

  @Override
  public Mono<Void> write(Consumer<OutputStream> consumer) {
    if (os == null) {
      mono = response.writeWith(Flux.create(fluxSink -> {
        try {
          os = new FluxSinkOutputStream(fluxSink, response);
          consumer.accept(os);
        } catch (Throwable e) {
          fluxSink.error(e);
        }
      }));
    } else {
      consumer.accept(os);
    }
    return mono;
  }

  @SneakyThrows
  @Override
  public void flush() {
    if (os != null) {
      os.flush();
    }
  }

  @AllArgsConstructor
  public static class FluxSinkOutputStream extends OutputStream {

    private FluxSink<DataBuffer> fluxSink;

    private ServerHttpResponse response;

    @Override
    public void write(byte[] b) throws IOException {
      writeSink(b);
    }

    @Override
    public void write(byte[] b, int off, int len) throws IOException {
      byte[] bytes = new byte[len];
      System.arraycopy(b, off, bytes, 0, len);
      writeSink(bytes);
    }

    @Override
    public void write(int b) throws IOException {
      writeSink((byte) b);
    }

    @Override
    public void flush() {
      fluxSink.complete();
    }

    public void writeSink(byte... bytes) {
      DataBuffer buffer = response.bufferFactory().wrap(bytes);
      fluxSink.next(buffer);
      //在这里可能有问题，但是目前没有没有需要释放的数据
      DataBufferUtils.release(buffer);
    }
  }
}

只要最后都是写byte[]就可以相互转化，只不过可能麻烦一点，需要用接口回调

将FluxSink伪装成一个OutputStream，写入时把byte[]转成DataBuffer 并调用next方法，最后在flush的时候调用complete方法就行了，完美

响应写入其实就是对输入输出流的处理了，正常情况下，我们会定义一个byte[]用来缓存读到的数据，所以我也不会固定这个缓存的大小而是提供了DownloadWriter可以自定义处理输入输出流，包括存在指定编码或是Range头的情况

/**
 * 具体操作 {@link InputStream} 和 {@link OutputStream} 的写入器。
 */
public interface DownloadWriter extends OrderProvider {

  /**
   * 该写入器是否支持写入。
   *
   * @param resource {@link Resource}
   * @param range    {@link Range}
   * @param context  {@link DownloadContext}
   * @return 如果支持则返回 true
   */
  boolean support(Resource resource, Range range, DownloadContext context);

  /**
   * 执行写入。
   *
   * @param is      {@link InputStream}
   * @param os      {@link OutputStream}
   * @param range   {@link Range}
   * @param charset {@link Charset}
   * @param length  总大小，可能为 null
   */
  default void write(InputStream is, OutputStream os, Range range, Charset charset, Long length) {
    write(is, os, range, charset, length, null);
  }

  /**
   * 执行写入。
   *
   * @param is       {@link InputStream}
   * @param os       {@link OutputStream}
   * @param range    {@link Range}
   * @param charset  {@link Charset}
   * @param length   总大小，可能为 null
   * @param callback 回调当前进度和增长的大小
   */
  void write(InputStream is, OutputStream os, Range range, Charset charset, Long length, Callback callback);

  /**
   * 进度回调。
   */
  interface Callback {

    /**
     * 回调进度。
     *
     * @param current  当前值
     * @param increase 增长值
     */
    void onWrite(long current, long increase);
  }
}

事件

当我把整个下载流程实现之后发现其实整个逻辑还是有点复杂的，所有得想个办法能监控整个下载流程

最开始我定义了几个监听器用来回调，但是并不好用，首先我们整个架构设计的是十分灵活可扩展的，而定义的监听器类型少而且不好扩展

当我们后续添加了其他的流程和步骤后，不得不新加几类监听器或是在原来的监听器类上添加方法，十分麻烦

所以我想到使用事件的方式能更加灵活的扩展，并定义了DownloadEventPublisher用于发布事件和DownloadEventListener用于监听事件，而且支持了Spring的事件监听方式

日志

基于上述的事件方式，我在此基础上实现了几种下载日志

  • 每个流程对应的日志
  • 加载进度更新，压缩进度更新，响应写入进度更新的日志
  • 时间花费的日志

这些日志由于比较详细的打印了整个下载流程的信息，还帮我发现了好多Bug

其他坑

最开始上下文的初始化和销毁各自对应了一个步骤分别位于最开始和最末尾，但是当我在webflux中写完响应后，发现上下文的销毁不会执行

于是我跟了下Spring的源码发现写入方法返回的是Mono.empty()，也就是说，当响应写入后就不会往下调用next方法了，所以在响应写入之后的步骤永远都不会被调用

最后就把上下文初始化和销毁单独出来了，并且在doAfterTerminate时调用销毁方法

结束

基本上的内容就是这样了，不过对于响应式这块的内容还是莫得不是很透，以及有部分操作符也不是很会用，但还是有了解到很多高级的用法

文章目录
  1. 1. 介绍
  2. 2. 思路
    1. 2.1. 基础
    2. 2.2. 架构
      1. 2.2.1. 下载上下文
      2. 2.2.2. 下载类型支持
      3. 2.2.3. 并发加载
      4. 2.2.4. 压缩
      5. 2.2.5. 响应写入
      6. 2.2.6. 事件
      7. 2.2.7. 日志
    3. 2.3. 其他坑
  3. 3. 结束