神奇注解：一鍵下載任意對象
介紹
下載功能應該是比較常見的功能了，雖然一個項目里面可能出現(xiàn)的不多，但是基本上每個項目都會有，而且有些下載功能其實還是比較繁雜的，倒不是難，而是麻煩。
所以結(jié)合之前的下載需求，我寫了一個庫來簡化下載功能的實現(xiàn)。
傳送門：https://github.com/Linyuzai/concept/wiki/Concept-Download。
如果我說現(xiàn)在只需要一個注解就能幫你下載任意的對象，是不是覺得非常的方便。
@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 地址。
現(xiàn)在需要導出所有設備二維碼圖片的壓縮包，圖片名稱需要用設備名稱加 .png 后綴，需求上來說并不難，但是實現(xiàn)著實有點麻煩。
- 首先需要將設備列表查出來
- 然后使用二維碼地址下載圖片并寫到本地緩存文件
- 在下載之前需要先判斷是否已經(jīng)存在緩存
- 下載時需要并發(fā)下載提升性能
- 等所有圖片下載結(jié)束后
- 再生成一個壓縮文件
- 然后再操作輸入輸出流寫到響應中
我實現(xiàn)了將近 200 行的代碼，真是又臭又長，一個下載功能咋能那么麻煩呢，于是我就想有沒有更簡單的方式。
我當時的需求很簡單，我想著我只要提供需要下載的數(shù)據(jù)，比如一個文件路徑，一個文件對象，一段字符串文本，一個http地址，或者混搭了前面所有類型的一個集合，甚至是我們自定義的某個類的實例，后面的事情我就不用管了。
文件路徑是一個文件還是一個目錄？字符串文本需要先寫入一個文本文件中？http資源如何下載到本地？多個文件怎么壓縮？最后怎么寫到響應中？我才不想花時間管這些。
比如就像我現(xiàn)在這個需求，我只要返回設備列表就行了，其他的事情我都不用管。
@Download(filename = "二維碼.zip")
@GetMapping("/download")
public List<Device> download() {
return deviceService.all();
}
public class Device {
//設備名稱
private String name;
//設備二維碼
//注解表示該http地址是需要下載的數(shù)據(jù)
@SourceObject
private String qrCodeUrl;
//注解表示文件名稱
@SourceName
public String getQrCodeName() {
return name + ".png";
}
//省略其他屬性方法
}通過在 Device 的字段上標注某些注解（或是實現(xiàn)某個接口）來指定文件名稱和文件地址。
如果能這樣實現(xiàn)，省時省心省力，又多了寫 199 行代碼的摸魚時間難道不香么。
思路
下面來講講這個庫的主要設計思路，以及中間遇到的坑。
其實基于一開始的設想，我覺得功能并沒有多復雜，于是就決定開肝。
只是萬萬沒想到實現(xiàn)起來比我想象的更復雜（這是后話了）。
基礎(chǔ)
首先整個庫基于響應式編程，但卻并不是完全意義上的響應式，只能說是Mono<InputStream>這樣的。。。奇怪組合？
為什么會這樣呢，很大的一個原因是由于需要兼容webmvc和webflux，導致我僅僅是將之前實現(xiàn)的InputStream方式重構(gòu)成了響應式，所以就出現(xiàn)了這樣的組合。
這也是我遇到的最大的一個坑，我先前已經(jīng)基本調(diào)通了基于Servlet的整個下載流程，然后就想著支持一下webflux。
大家都知道webmvc中，我們可以通過RequestContextHolder來獲得請求和響應對象，但是在webflux中就不行了，當然我們可以在方法參數(shù)中注入。
@Download(source = "classpath:/download/README.txt")
@GetMapping("/classpath")
public void classpath(ServerHttpResponse response) {
}結(jié)合Spring自帶的注入功能，我們就可以通過AOP拿到響應的入?yún)⒘?，但是總覺得這樣寫有點多余，強迫癥表示不能忍。
有什么辦法既能把用不到的入?yún)⒏傻?，又能拿到響應對象呢，在網(wǎng)上找到了一種實現(xiàn)方式。
/**
* 用于設置當前的請求和響應。
*
* @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的非阻塞線程，如果調(diào)用該方法會直接拋出異常。
所以就沒有任何辦法了，只能將之前代碼基于響應式重構(gòu)。
架構(gòu)
接下來說說整體架構(gòu)：
圖片
對于一個下載請求，我們可以分成幾個步驟，以下載多個文件的壓縮包為例：
- 首先我們一般是得到多個文件的路徑或?qū)腇ile對象。
- 然后將這些文件壓縮生成一個壓縮文件。
- 最后將壓縮文件寫入到響應中。
但是對于我上面描述的需求，一開始就不是文件路徑或?qū)ο罅耍且粋€http地址，然后在壓縮之前還需要多一個步驟，需要先將圖片下載下來。
那么對于各種各樣的需求我們可能需要在當前步驟中的任意位置添加額外的步驟，所以我參考了Spring Cloud Gateway 攔截鏈的實現(xiàn)方式。
/**
* 下載處理器。
*/
public interface DownloadHandler extends OrderProvider {
/**
* 執(zhí)行處理。
*
* @param context {@link DownloadContext}
* @param chain {@link DownloadHandlerChain}
*/
Mono<Void> handle(DownloadContext context, DownloadHandlerChain chain);
}
/**
* 下載處理鏈。
*/
public interface DownloadHandlerChain {
/**
* 調(diào)度下一個下載處理器。
*
* @param context {@link DownloadContext}
*/
Mono<Void> next(DownloadContext context);
}這樣每個步驟就可以單獨實現(xiàn)一個DownloadHandler，步驟與步驟之間可以任意的組合添加。
下載上下文
在此基礎(chǔ)上使用一個貫穿整個流程的上下文DownloadContext，方便共享和傳遞步驟之間的中間結(jié)果。
對于上下文DownloadContext也提供了DownloadContextFactory可以用于自定義上下文。
同時提供了DownloadContextInitializer和DownloadContextDestroyer用于在上下文初始化和銷毀時擴展自己的邏輯。
下載類型支持
我們需要下載的數(shù)據(jù)的類型是不固定的，比如有文件，有http地址，也會有之前我希望的自定義的類的實例。
所以我將所有的下載對象抽象成了Source，表示一個下載源，這樣文件可以實現(xiàn)為FileSource，http地址可以實現(xiàn)為HttpSource，然后通過對應的SourceFactory來匹配創(chuàng)建。
比如FileSourceFactory可以匹配File并且創(chuàng)建FileSource，HttpSourceFactory可以匹配http://前綴并且創(chuàng)建HttpSource。
/**
* {@link Source} 工廠。
*/
public interface SourceFactory extends OrderProvider {
/**
* 是否支持需要下載的原始數(shù)據(jù)對象。
*
* @param source 需要下載的原始數(shù)據(jù)對象
* @param context {@link DownloadContext}
* @return 如果支持則返回 true
*/
boolean support(Object source, DownloadContext context);
/**
* 創(chuàng)建。
*
* @param source 需要下載的原始數(shù)據(jù)對象
* @param context {@link DownloadContext}
* @return 創(chuàng)建的 {@link Source}
*/
Source create(Object source, DownloadContext context);
}那么對于我們自定義的類要怎么支持呢，之前提到可以在類上標注注解或是實現(xiàn)特定的接口，那么就用我實現(xiàn)的注解的方式來大概講一講吧！
其實邏輯很簡單，只要能熟練的運用反射就完全沒問題，我們再來看一看用法。
@Download(filename = "二維碼.zip")
@GetMapping("/download")
public List<Device> download() {
return deviceService.all();
}
public class Device {
//設備名稱
private String name;
//設備二維碼
//注解表示該http地址是需要下載的數(shù)據(jù)
@SourceObject
private String qrCodeUrl;
//注解表示文件名稱
@SourceName
public String getQrCodeName() {
return name + ".png";
}
//省略其他屬性方法
}首先我定義了一個注解@SourceModel標注在類上表示需要被解析，然后定義了一個@SourceObject注解標注在需要下載的字段（或方法）上，這樣我們就可以通過反射拿到這個字段（或方法）的值。
基于當前支持的SourceFactory就能創(chuàng)建出對應的Source，接下來使用@SourceName指定名稱，也同樣可以通過反射獲得這個方法（或字段）的值并依舊通過反射設置到創(chuàng)建出來的Source上。
這樣就能非常靈活的支持任意的對象類型了。
并發(fā)加載
對于像http這種網(wǎng)絡資源，我們需要先并發(fā)加載（多個文件時）到本地的內(nèi)存中或是緩存文件中來提升我們的處理效率。
當然我可以直接定死一個線程池來執(zhí)行，但是每個機器每個項目甚至每個需求對于并發(fā)的要求和資源的分配都不一樣。
所以我提供了SourceLoader來支持自定義的加載邏輯，你甚至可以一部分用線程池，一部分用協(xié)程，剩下一部分不加載。
/**
* {@link Source} 加載器。
*
* @see DefaultSourceLoader
* @see SchedulerSourceLoader
*/
public interface SourceLoader {
/**
* 執(zhí)行加載。
*
* @param source {@link Source}
* @param context {@link DownloadContext}
* @return 加載后的 {@link Source}
*/
Mono<Source> load(Source source, DownloadContext context);
}壓縮
當我們加載完之后就可以執(zhí)行壓縮了，同樣的我定義了一個類Compression作為壓縮對象的抽象。
一般來說，我們會先在本地創(chuàng)建一個緩存文件，然后將壓縮后的數(shù)據(jù)寫入到緩存文件中。
不過我每次都很討厭在配置文件中配置各種各樣的路徑，所以在壓縮時支持內(nèi)存壓縮，當然如果文件比較大還是老老實實生成一個緩存文件。
對于壓縮格式也提供了可以完全自定義的SourceCompressor接口，你想自己實現(xiàn)一個壓縮協(xié)議都沒有問題。
/**
* {@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());
}
/**
* 如果支持對應的格式就會調(diào)用該方法執(zhí)行壓縮。
*
* @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。
但是問題又出現(xiàn)了，下面是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);
//在這里可能有問題，但是目前沒有沒有需要釋放的數(shù)據(jù)
DataBufferUtils.release(buffer);
}
}
}只要最后都是寫byte[]就可以相互轉(zhuǎn)化，只不過可能麻煩一點，需要用接口回調(diào)。
將FluxSink偽裝成一個OutputStream，寫入時把byte[]轉(zhuǎn)成DataBuffer 并調(diào)用next方法，最后在flush的時候調(diào)用complete方法就行了，完美。
響應寫入其實就是對輸入輸出流的處理了，正常情況下，我們會定義一個byte[]用來緩存讀到的數(shù)據(jù)，所以我也不會固定這個緩存的大小而是提供了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);
/**
* 執(zhí)行寫入。
*
* @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);
}
/**
* 執(zhí)行寫入。
*
* @param is {@link InputStream}
* @param os {@link OutputStream}
* @param range {@link Range}
* @param charset {@link Charset}
* @param length 總大小，可能為 null
* @param callback 回調(diào)當前進度和增長的大小
*/
void write(InputStream is, OutputStream os, Range range, Charset charset, Long length, Callback callback);
/**
* 進度回調(diào)。
*/
interface Callback {
/**
* 回調(diào)進度。
*
* @param current 當前值
* @param increase 增長值
*/
void onWrite(long current, long increase);
}
}事件
當我把整個下載流程實現(xiàn)之后發(fā)現(xiàn)其實整個邏輯還是有點復雜的，所有得想個辦法能監(jiān)控整個下載流程。
最開始我定義了幾個監(jiān)聽器用來回調(diào)，但是并不好用，首先我們整個架構(gòu)設計的是十分靈活可擴展的，而定義的監(jiān)聽器類型少而且不好擴展。
當我們后續(xù)添加了其他的流程和步驟后，不得不新加幾類監(jiān)聽器或是在原來的監(jiān)聽器類上添加方法，十分麻煩。
所以我想到使用事件的方式能更加靈活的擴展，并定義了DownloadEventPublisher用于發(fā)布事件和DownloadEventListener用于監(jiān)聽事件，而且支持了Spring的事件監(jiān)聽方式。
日志
基于上述的事件方式，我在此基礎(chǔ)上實現(xiàn)了幾種下載日志。
- 每個流程對應的日志。
- 加載進度更新，壓縮進度更新，響應寫入進度更新的日志。
- 時間花費的日志。
這些日志由于比較詳細的打印了整個下載流程的信息，還幫我發(fā)現(xiàn)了好多Bug。
其他坑
最開始上下文的初始化和銷毀各自對應了一個步驟分別位于最開始和最末尾，但是當我在webflux中寫完響應后，發(fā)現(xiàn)上下文的銷毀不會執(zhí)行。
于是我跟了下Spring的源碼發(fā)現(xiàn)寫入方法返回的是Mono.empty()，也就是說，當響應寫入后就不會往下調(diào)用next方法了，所以在響應寫入之后的步驟永遠都不會被調(diào)用。
最后就把上下文初始化和銷毀單獨出來了，并且在doAfterTerminate時調(diào)用銷毀方法。
