如何使用vertx反向代理处理大文件的下载

野性酷女 2023-06-15 13:26 9阅读 0赞

### 前言 ###

由于我们的api-gateway使用vertx web开发，而大文件的下载在通过网关处理时,会超时并且严重拖慢网关的处理。所以需要单独优化处理了下载接口，让网关可以将后端传过来的buffer直接发送给客户端，而不需要等待网关下载到网关的服务器上，再转发给客户端（这也是大文件下载客户端会超时的一部分原因）

### 问题所在 ###

webclient对body的处理：  
vertx webclient默认下载文件时，是先下载到同级目录下的file-uploads文件夹下，下载全部完成后才会触发handler回调，而大文件时，可能会持续几分钟甚至更长，而这个时候持有的客户端请求八成已经超时断开了。

### 解决方案 ###

webclient其实对body的处理是可以配置的，可以配置成pipe模式，并传入自定义的writeStream来实现自定义的流处理方案，而在routingContext中也是可以拿到前端的netSocket的，也就可以拿到客户端的writeStream。所以我们的方案就是，自定义一个GatewayWriteStream,然后把客户端的netsocket交给这个GatewayWriteStream。具体的代码如下：

import io.vertx.core.*;
    import io.vertx.core.buffer.Buffer;
    import io.vertx.core.http.HttpHeaders;
    import io.vertx.core.logging.Logger;
    import io.vertx.core.logging.LoggerFactory;
    import io.vertx.core.net.NetSocket;
    import io.vertx.core.streams.WriteStream;
    
    import java.util.Objects;
    
    /** * @author liwei * @title: GatewayWriteStream * @projectName moho * @description: 下载文件的流处理 * @date 2019-11-22 13:30 */
    public class GatewayWriteStream implements WriteStream<Buffer> { 
    
        private NetSocket netSocket;
    
        private Handler<Throwable> exceptionHandler;
    
        private int maxWrites = 128 * 1024;    // TODO - we should tune this for best pernformace
    
        private long writesOutstanding;
    
        private Runnable closedDeferred;
    
        private boolean closed;
    
        private final Vertx vertx;
    
        private final Context context;
    
        private int lwm = maxWrites / 2;
    
        private Handler<Void> drainHandler;
    
        private Logger log;
    
    
        public GatewayWriteStream(NetSocket netSocket, Vertx vertx,String length) { 
            Objects.requireNonNull(netSocket, "NetSocket");
            this.netSocket = netSocket;
            //返回响应头，来通知客户端文件大小
            String header="HTTP/1.0 200 \n"+
                    HttpHeaders.CONTENT_LENGTH.toString()+": "+length+" \n\n";
            Buffer buffer=Buffer.buffer(header);
            this.netSocket.write(buffer);
            this.vertx = vertx;
            this.context=vertx.getOrCreateContext();
            this.log= LoggerFactory.getLogger(this.getClass());
        }
    
        @Override
        public WriteStream<Buffer> exceptionHandler(Handler<Throwable> handler) { 
            check();
            this.exceptionHandler = handler;;
            return this;
        }
    
        @Override
        public WriteStream<Buffer> write(Buffer buffer) { 
            return write(buffer, null);
        }
    
        @Override
        public synchronized WriteStream<Buffer> write(Buffer buffer, Handler<AsyncResult<Void>> handler) { 
            doWrite(buffer, handler);    
            return this;
        }
    
        @Override
        public void end() { 
            netSocket.end();
        }
    
        @Override
        public void end(Handler<AsyncResult<Void>> handler) { 
            netSocket.end(handler);
        }
    
        @Override
        public WriteStream<Buffer> setWriteQueueMaxSize(int maxSize) { 
            netSocket.setWriteQueueMaxSize(maxSize);
            return this;
        }
    
        @Override
        public boolean writeQueueFull() { 
            return netSocket.writeQueueFull();
        }
    
        @Override
        public WriteStream<Buffer> drainHandler(Handler<Void> handler) { 
            check();
            this.drainHandler = handler;
            checkDrained();
            return this;
        }
    
        private synchronized WriteStream<Buffer> doWrite(Buffer buffer, Handler<AsyncResult<Void>> handler) { 
            Objects.requireNonNull(buffer, "buffer");
            check();
            Handler<AsyncResult<Void>> wrapped = ar -> { 
                if (ar.succeeded()) { 
                    checkContext();
                    Runnable action;
                    synchronized (GatewayWriteStream.this) { 
                        if (writesOutstanding == 0 && closedDeferred != null) { 
                            action = closedDeferred;
                        } else { 
                            action = this::checkDrained;
                        }
                    }
                    action.run();
                    if (handler != null) { 
                        handler.handle(ar);
                    }
                } else { 
                    if (handler != null) { 
                        handler.handle(ar);
                    } else { 
                        handleException(ar.cause());
                    }
                }
            };
    
            doWriteBuffer(buffer,buffer.length(),wrapped);
    
            return this;
        }
    
        private void doWriteBuffer(Buffer buff, long toWrite, Handler<AsyncResult<Void>> handler) { 
            if (toWrite > 0) { 
                synchronized (this) { 
                    writesOutstanding += toWrite;
                }
                writeInternal(buff, handler);
            } else { 
                handler.handle(Future.succeededFuture());
            }
        }
    
    
    
        private void writeInternal(Buffer buff, Handler<AsyncResult<Void>> handler) { 
            netSocket.write(buff,as->{ 
                if(as.succeeded()){ 
                    synchronized (GatewayWriteStream.this) { 
                        writesOutstanding -= buff.getByteBuf().nioBuffer().limit();
                    }
                    handler.handle(Future.succeededFuture());
                }
            });
        }
    
        private synchronized void closeInternal(Handler<AsyncResult<Void>> handler) { 
            check();
    
            closed = true;
    
            if (writesOutstanding == 0) { 
                doClose(handler);
            } else { 
                closedDeferred = () -> doClose(handler);
            }
        }
    
        private void check() { 
            checkClosed();
        }
    
        private void checkClosed() { 
            if (closed) { 
                throw new IllegalStateException("File handle is closed");
            }
        }
    
        private void doClose(Handler<AsyncResult<Void>> handler) { 
            Context handlerContext = vertx.getOrCreateContext();
            handlerContext.executeBlocking(res -> { 
                netSocket.end();
                res.complete(null);
    
            }, handler);
        }
    
        private void checkContext() { 
            if (!vertx.getOrCreateContext().equals(context)) { 
                throw new IllegalStateException("AsyncFile must only be used in the context that created it, expected: "
                        + context + " actual " + vertx.getOrCreateContext());
            }
        }
    
        private synchronized void checkDrained() { 
            if (drainHandler != null && writesOutstanding <= lwm) { 
                Handler<Void> handler = drainHandler;
                drainHandler = null;
                handler.handle(null);
            }
        }
    
        private void handleException(Throwable t) { 
            if (exceptionHandler != null && t instanceof Exception) { 
                exceptionHandler.handle(t);
            } else { 
                log.error("Unhandled exception", t);
    
            }
        }
    }

注：这里的代码借鉴了vertx官方的AsyncFileImpl,其中很多buffer长度和位置的记录是可以省略的，只不过我这里为了后续其他功能保留了下来，不需要可以自行更改。

router的handler中的代码，类似如下：

private void downLoad(TargetInfo targetInfo, RoutingContext context, String length) { 
            SocketAddress socketAddress = SocketAddress.inetSocketAddress(targetInfo.getPort(), targetInfo.getHost());
            GatewayWriteStream writeStream = new GatewayWriteStream(context.request().netSocket(), context.vertx(), length);
            webClient.request(HttpMethod.GET, socketAddress, targetInfo.getRemoteUri()).
                    as(BodyCodec.pipe(writeStream)).send(v -> { 
    
            });
        }

而传入的length是文件的大小，他的获取方式是在真正请求之前，先向后端发起一次head请求，返回的headers中的CONTENT\_LENGTH就是文件的大小（这是我目前的处理方式，但这样会造成每次需要请求两次后端，如果有更好的解决方式我会更新，如果你发现了也请留言告诉我）

#### 扩展 ####

与其对应的应该还有网关上传功能，我目前没有做，不过可以想象的到方式大概如下：

1.  配置router的bodyhandler对body的处理方式为pipe,然后传入自定义的writestream,这样我们就拿到了客户端上传的流
2.  自定义实现readstream接口，将1中的写流传入进来，然后调用webclient.sendStream，将自定义的readstream传入应该就可以了。
3.  如果有谁做了，可以贴个链接分享给我

### 2019-12-11补充： ###

上传大文件的转发：

实现思路和之前想的差不多，只不过不能使用webclient，因为它封装时候已经把buffer的handler隐藏掉了，只能通过httpclient来处理。不多说看代码：

HttpClient client = vertx.createHttpClient(new HttpClientOptions());
                HttpClientRequest c_req=client.request(VertHttpRequestWrapper.transMethod(request.getMethod()), SocketAddress.inetSocketAddress(targetInfo.getPort(),targetInfo.getHost())
                        ,targetInfo.getPort(),targetInfo.getHost(),targetInfo.getRemoteUri(), res->{ 
                    context.response().setChunked(true);
                    context.response().setStatusCode(res.statusCode());
                    context.response().headers().setAll(res.headers());
                    res.handler(data->{ 
                       req.response().write(data);
                    });
                    res.endHandler((v)->req.response().end());
                });
                c_req.setChunked(true);
                c_req.headers().setAll(context.request().headers());
                req.handler(c_req::write);
                req.endHandler((v) -> c_req.end());

这里的处理就是把客户端发送的buffer流直接转发给了后端服务器，这样做存在的问题是，不能对Body进行任何校验，网关就起到了反向代理的作用,并且如果不在前边对请求增加校验，很容易因为配置问题导致网关变成一个简单的反向代理服务。  
如何校验是上传：

String contentType = context.request().getHeader(HttpHeaders.CONTENT_TYPE);
                if(contentType==null){ 
                    context.fail(RouterException.e(RouterCode.REQUEST_ERROR.getCode(),
                            RouterCode.REQUEST_ERROR.getMessage() + ": "+ "该请求并不是上传请求,请核实请求头",
                            RouterCode.REQUEST_ERROR.getStatus()));
                    return;
                }
                String lowerCaseContentType = contentType.toLowerCase();
                boolean isMultipart = lowerCaseContentType.startsWith(HttpHeaderValues.MULTIPART_FORM_DATA.toString());
                boolean isUrlEncoded = lowerCaseContentType.startsWith(HttpHeaderValues.APPLICATION_X_WWW_FORM_URLENCODED.toString());
                if(!isMultipart&&!isUrlEncoded){ 
                    context.fail(RouterException.e(RouterCode.REQUEST_ERROR.getCode(),
                            RouterCode.REQUEST_ERROR.getMessage() + ": "+ "该请求并不是上传请求,请核实请求头",
                            RouterCode.REQUEST_ERROR.getStatus()));
                    return;
                }