本人在中间件研发组(主要开发RPC),近期遇到一个需求:RPC基于http协议通过netty支持文件上传下载
经过一系列的资料查找学习,终于实现了该功能
通过netty实现文件上传下载,主要在编解码时处理,具体的代码如下:
①文件上传
@Override public RPCRequest doDecodeRequest(FullHttpRequest request) { RPCRequest rpcRequest = new RPCRequest(); String biz_prefix = BIZ_PREFIX; String rpc_prefix = RPC_PREFIX; if("true".equals(request.headers().get(RPCParamType.header_compatible.getName()))){ biz_prefix = ""; rpc_prefix = ""; } for (Entry<String, String> entry : request.headers().entries()) { if(entry.getKey().startsWith(rpc_prefix)){ rpcRequest.addRpcHeader(entry.getKey().substring(rpc_prefix.length()), entry.getValue()); } if(entry.getKey().startsWith(biz_prefix)){ rpcRequest.addBizHeader(entry.getKey().substring(biz_prefix.length()), entry.getValue()); } } QueryStringDecoder decoderQuery = new QueryStringDecoder(request.uri()); // serviceId String serviceId = decoderQuery.path(); if (serviceId.startsWith(RPCConstant.PATH_SEPARATOR)) { serviceId = serviceId.substring(1); } if (serviceId.endsWith(RPCConstant.PATH_SEPARATOR)) { serviceId = serviceId.substring(0, serviceId.length() - 1); } rpcRequest.addRpcHeader(RPCParamType.service_id.getName(), serviceId); // requestId String requestId = request.headers().get(rpc_prefix + RPCConstant.RequestId); if (null == requestId || "".equals(requestId)) { requestId = "0"; } rpcRequest.setId(Long.valueOf(requestId)); // add by wangzp in 2017/3/30 if (HttpMethod.GET == request.method()) { rpcRequest.addBizHeader("HTTP_METHOD", "GET"); Map<String, String> map = new HashMap<String, String>(); for (Map.Entry<String, List<String>> paramsEntry : decoderQuery.parameters().entrySet()) { map.put(paramsEntry.getKey(), paramsEntry.getValue().get(0)); rpcRequest.getBizHeader().put(paramsEntry.getKey(), paramsEntry.getValue().get(0)); } //TODO 需要引入序列化机制,否则data部分找不到相匹配的序列化 //TODO 暂时用json rpcRequest.setData(JSON.toJSONString(map)); } else { try { decoder = new HttpPostRequestDecoder(factory, request); readingChunks = HttpUtil.isTransferEncodingChunked(request); if(decoder != null && readingChunks){ HttpContent chunk = (HttpContent) request; try{ decoder.offer(chunk); } catch (ErrorDataDecoderException e) { decoder.destroy(); } } File file = readHttpDataChunkByChunk(); //从解码器decoder中读出数据 rpcRequest.addBizHeader(RPCParamType.upload_file.getName(), file.getPath()); } catch (Exception e) { decoder.destroy(); } } return rpcRequest; }
②文件下载
@Override public FullHttpResponse doEncodeResponse(RPCResponse response) { FullHttpResponse httpResponse ; String biz_prefix = BIZ_PREFIX; String rpc_prefix = RPC_PREFIX; if(response.getBizHeader(RPCParamType.header_compatible.getName()) == null ? RPCParamType.header_compatible.getBooleanValue() : Boolean.parseBoolean(response.getBizHeader(RPCParamType.header_compatible.getName()))){ biz_prefix = ""; rpc_prefix = ""; } String downloadFile = response.getBizHeader(RPCParamType.download_file.getName()); if(!StringUtils.isEmpty(downloadFile)){ File file = new File(downloadFile); String fileName = file.getName(); ByteBuf buffer; try { buffer = Unpooled.copiedBuffer(Files.readAllBytes(file.toPath())); httpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,HttpResponseStatus.OK,buffer); } catch (IOException e) { throw new RuntimeException(fileName + "is not exist"); } httpResponse.headers().add(HttpHeaderNames.CONTENT_TYPE, "application/octet-stream; charset=utf-8"); try { httpResponse.headers().add(HttpHeaderNames.CONTENT_DISPOSITION, "attachment; filename="" + new String(fileName.getBytes("gb2312"),"ISO-8859-1") + """); } catch (UnsupportedEncodingException e) { throw new RuntimeException(fileName + "download fail"); } }else{ //serialize byte[] body = (byte[]) SerializerUtils.serialize(getProtocolUrl().getParameter(RPCParamType.serializer_name.getName(), RPCParamType.serializer_name.getValue()), response.getData()); String reasonPhrase = response.getBizHeader(RPCParamType.http_reasonPhrase.getName()); response.getBizHeader().remove(RPCParamType.http_reasonPhrase.getName()); //reasonPhrase不需要放入RPCResponse中传输 httpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(Integer.valueOf(response.getBizHeader(RPCConstant.Code) == null ? response.getRpcHeader(RPCConstant.Code) : response.getBizHeader(RPCConstant.Code)), reasonPhrase == null ? RPCParamType.http_reasonPhrase.getValue() : reasonPhrase), Unpooled.wrappedBuffer(body)); } for (Entry<String, String> entry : response.getBizHeader().entrySet()) { httpResponse.headers().add(biz_prefix + entry.getKey(), entry.getValue()); } for (Entry<String, String> entry : response.getRpcHeader().entrySet()) { httpResponse.headers().add(rpc_prefix + entry.getKey(), entry.getValue()); } httpResponse.headers().add(rpc_prefix + RPCConstant.RequestId, response.getRequestId()); httpResponse.headers().add(CLZNAME, response.getClzName()); httpResponse.headers().add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE); httpResponse.headers().add(HttpHeaderNames.CONTENT_LENGTH, httpResponse.content().readableBytes()); return httpResponse; }
③其他辅助方法
public class HTTPExCodec extends HTTPCodec { private boolean readingChunks; private static final HttpDataFactory factory = new DefaultHttpDataFactory(DefaultHttpDataFactory.MINSIZE) ; private HttpPostRequestDecoder decoder; //此处包含上面两个方法 。。。。 private File readHttpDataChunkByChunk() { try { while (decoder.hasNext()) { InterfaceHttpData data = decoder.next(); if (data != null) { try { File file = writeHttpData(data); return file; } finally { data.release(); } } } } catch (EndOfDataDecoderException e1) { throw new RuntimeException("write file error"); } return null; } private File writeHttpData(InterfaceHttpData data) { String uploadFileName = getUploadFileName(data); FileUpload fileUpload = (FileUpload) data; if (fileUpload.isCompleted()) { File dir = new File(System.getProperty("user.dir") + File.separator); if (!dir.exists()) { dir.mkdir(); } File dest = new File(dir, uploadFileName); try { fileUpload.renameTo(dest); return dest; } catch (IOException e) { throw new RuntimeException("write file error"); } } return null; } private String getUploadFileName(InterfaceHttpData data) { String content = data.toString(); String temp = content.substring(0, content.indexOf(" ")); content = temp.substring(temp.lastIndexOf("=") + 2, temp.lastIndexOf(""")); return content; } }