如何进行大文件传输？

news/2024/11/18 14:51:09/

本文首发微信公众号：码上观世界

网络文件传输的应用场景很多，如网络聊天的点对点传输、文件同步网盘的上传与下载、文件上传到分布式文件存储器等，其传输速度主要受限于网络带宽、存储器大小、CPU处理速度以及磁盘读写速度，尤其是网络带宽。本文主要讨论通常情况下数十GB规模大小的文件传输的优化方式，对于更大规模的文件容量建议考虑人工硬盘运输，毕竟基于公路运输的方式不仅带宽大而且成本低。

文件传输涉及到客户端、中间网络和服务器，常用的传输协议有HTTP(s)、(S)FTP和TCP（UDP）协议等，对于客户端用户来讲，能够起作用的地方不大，所以本文就两种基本的场景来讨论文件传输在客户端的优化方式：基于HTTP协议的非结构化文件传输和基于TCP协议的结构化文件传输。

基于HTTP协议的非结构化文件传输

最常用的文件上传是基于HTTP POST。观察浏览器的请求头数据可知，文件的二进制数据被置于请求body里面，也就是说在上传文件过程中，客户端是一次性将文件内容加载到内存，如果文件过大，浏览器很可能会崩溃，加上HTTP请求连接本身有超时时间限制，所以这种方式不适合传输大文件。

所以一种自然的方式就是手写符合规范的HTTP协议跟服务端通信：

上面的示例代码相比通过浏览器上传文件方式显得自由度更大，但是问题也更多，比如OutputStream将数据写入到PosterOutputStream内部缓冲区，而该缓冲区只有当调用HttpURLConnection的getInputStream方法之后才会发送到Socket流中。所以当文件过大（也许几十MB）就会导致内存溢出，即使通过调用flush方法也无济于事，因为PosterOutputStream的flush方法是空操作，什么都不干！幸运的是HttpURLConnection提供的setFixedLengthStreamingMode方法能够获取到自动刷新流缓存的StreamingOutputStream。虽然这种方式能够解决问题，但是还可能会遇到其他大大小小的坑，而且上述方式还是过于原始，使用Apache HttpClient能够轻易实现上述功能：

HttpClientBuilder httpClientBuilder = HttpClientBuilder.create();
httpClientBuilder.setDefaultCredentialsProvider(credsProvider);
RequestConfig requestConfig = RequestConfig.custom().setCookieSpec(CookieSpecs.DEFAULT).build();
CloseableHttpClient httpClient = httpClientBuilder.build();
File file = new File(filePath);
HttpPut httpPut = new HttpPut(url);
FileEntity fileEntity = new FileEntity(file);
httpPut.setEntity(fileEntity);
FileInputStream fileInputStream = new FileInputStream(file);
InputStreamEntity reqEntity = new InputStreamEntity(fileInputStream, file.length());
//post.setEntity(reqEntity);
HttpResponse response = httpClient.execute(httpPut);
String content = EntityUtils.toString(response.getEntity());

示例代码中，HttpClient帮我们封装了协议相关的所有内容。对于文件传输FileEntity 和InputStreamEntity 都可以使用，不同的是，InputStreamEntity 用了流传输的方式，我们需要做的是就是验证这两种方式是否存在文件过大导致的内存溢出问题。先看FileEntity ，直接翻到代码DefaultBHttpClientConnection ：

class DefaultBHttpClientConnection extends BHttpConnectionBase{
......
public void sendRequestEntity(final HttpEntityEnclosingRequest request)
throws HttpException, IOException {Args.notNull(request, "HTTP request");ensureOpen();final HttpEntity entity = request.getEntity();if (entity == null) {return;}final OutputStream outstream = prepareOutput(request);entity.writeTo(outstream);outstream.close();}
......
}
class FileEntity{
......
public void writeTo(final OutputStream outstream) throws IOException {
Args.notNull(outstream, "Output stream");final InputStream instream = new FileInputStream(this.file);try {final byte[] tmp = new byte[OUTPUT_BUFFER_SIZE];int l;while ((l = instream.read(tmp)) != -1) {outstream.write(tmp, 0, l);}outstream.flush();} finally {instream.close();}
}
......
}

再看看InputStreamEntity：

class InputStreamEntity{
......
public void writeTo(final OutputStream outstream) throws IOException {
Args.notNull(outstream, "Output stream");final InputStream instream = this.content;try {final byte[] buffer = new byte[OUTPUT_BUFFER_SIZE];int l;if (this.length < 0) {// consume until EOFwhile ((l = instream.read(buffer)) != -1) {outstream.write(buffer, 0, l);}} else {// consume no more than lengthlong remaining = this.length;while (remaining > 0) {l = instream.read(buffer, 0, (int)Math.min(OUTPUT_BUFFER_SIZE, remaining));if (l == -1) {break;}outstream.write(buffer, 0, l);remaining -= l;}}} finally {instream.close();}
}
......
}

可见FileEntity 和InputStreamEntity使用了相同的outstream，其生成方式为：

class BHttpConnectionBase{
......protected OutputStream prepareOutput(final HttpMessage message) throws HttpException {final long len = this.outgoingContentStrategy.determineLength(message);return createOutputStream(len, this.outbuffer);}protected OutputStream createOutputStream(final long len,final SessionOutputBuffer outbuffer) {if (len == ContentLengthStrategy.CHUNKED) {return new ChunkedOutputStream(2048, outbuffer);} else if (len == ContentLengthStrategy.IDENTITY) {return new IdentityOutputStream(outbuffer);} else {return new ContentLengthOutputStream(outbuffer, len);}}
......
}

这里以ContentLengthOutputStream为例来看数据是如何发送到Socket流中的：

class ContentLengthOutputStream{private final SessionOutputBuffer out;
......public void write(final byte[] b, final int off, final int len) throws IOException {if (this.closed) {throw new IOException("Attempted write to closed stream.");}if (this.total < this.contentLength) {final long max = this.contentLength - this.total;int chunk = len;if (chunk > max) {chunk = (int) max;}this.out.write(b, off, chunk);this.total += chunk;}
}
......
}
class SessionOutputBufferImpl{private OutputStream outstream;
......public void write(final byte[] b, final int off, final int len) throws IOException {if (b == null) {return;}// Do not want to buffer large-ish chunks// if the byte array is larger then MIN_CHUNK_LIMIT// write it directly to the output streamif (len > this.fragementSizeHint || len > this.buffer.capacity()) {// flush the bufferflushBuffer();// write directly to the out streamstreamWrite(b, off, len);this.metrics.incrementBytesTransferred(len);} else {// Do not let the buffer grow unnecessarilyfinal int freecapacity = this.buffer.capacity() - this.buffer.length();if (len > freecapacity) {// flush the bufferflushBuffer();}// bufferthis.buffer.append(b, off, len);}
}
private void flushBuffer() throws IOException {final int len = this.buffer.length();if (len > 0) {streamWrite(this.buffer.buffer(), 0, len);this.buffer.clear();this.metrics.incrementBytesTransferred(len);}
}
private void streamWrite(final byte[] b, final int off, final int len) throws IOException {Asserts.notNull(outstream, "Output stream");this.outstream.write(b, off, len);
}
......
}
class SocketOutputStream {
......public void write(byte b[], int off, int len) throws IOException {socketWrite(b, off, len);}
......
}

通过上面关键代码可见，不管用哪一种Entity，当缓冲区满了就自动flush到Socket，理论上都可以进行大文件传输，只要超时时间允许，两者并没有什么特别的不同。

基于TCP协议的结构化文件传输

基于HTTP协议的文件传输，虽然通过流的方式能解决大文件传输问题，但是基于应用层协议毕竟效率不到，时间消耗仍是个大问题，尽管可以通过文件拆分，并行处理，但需要服务器端的配合才能完成（比如将小文件还原，断点续传等）。这里讨论的多文件传输到分布式系统不需要对服务端再做改造就能直接使用，天然具备并行处理能力。对于结构化文件传输的使用场景多用于数据迁移，比如从数据库系统或者文件系统传输到大数据存储计算平台。这里以将本地的CSV文件上传到HDFS为例，需要解决的是如何对文件拆分。虽然对非结构化，半结构化文件因为涉及到分隔符问题，对于文件拆分有点儿难度，但对规范化格式的文件，问题倒不大，但考虑让问题描述更简洁，这里不考虑文件拆分，只考虑一个文件（比如文件夹下已经拆分后的某个文件）的传输问题。该问题模型可以描述为：

引入Channel是为了解决File和HDFS存取速率不匹配的问题，通过Channel连接File读过程和HDFS写过程：当Channel缓存满的时候，File等待HDFS读取之后再开始写入Channel，HDFS读取之后File再写入Channel，两者通过信号量机制协调，HDFS每次写入都是一个独立的文件。关键代码实现如下：

File端读取数据到Channel：

public void readCSV(String filePath, String fieldDelimiter) {
......BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, "UTF-8"), 8192);CsvReader csvReader = new CsvReader(reader);csvReader.setDelimiter(fieldDelimiter.charAt(0));String[] parseRows;while ((parseRows = splitBufferedReader(csvReader)) != null) {//Record 为文件中一行数据记录，由Column组成Record record = createRecord(parseRows);this.buffer.add(record);if (this.buffer.size() >= MemoryChannel.bufferSize) {this.channel.pushAll(this.buffer);this.buffer.clear();}}this.channel.pushAll(this.buffer);this.buffer.clear();
......}
//基于内存的Channel实现
class  MemoryChannel{private ArrayBlockingQueue<Record> queue;private ReentrantLock lock;private Condition notInsufficient, notEmpty;
......//将File读取端将记录push到Channelpublic void pushAll(final Collection<Record> rs) {Validate.notNull(rs);Validate.noNullElements(rs);try {lock.lockInterruptibly();while (!this.queue.isEmpty()) {notInsufficient.await(200L, TimeUnit.MILLISECONDS);}this.queue.addAll(rs);notEmpty.signalAll();} catch (InterruptedException e) {throw new RuntimeException("pushAll", e);} finally {lock.unlock();}}......
}
class HdfsWriteService{
......
public void writeFile(String fieldDelimiter) {FileOutputFormat outFormat = new TextOutputFormat();outFormat.setOutputPath(jobConf, outputPath);outFormat.setWorkOutputPath(jobConf, outputPath);List<Record> recordList= new ArrayList(MemoryChannel.bufferSize);this.channel.pullAll(recordList);RecordWriter writer = outFormat.getRecordWriter(fileSystem, jobConf, outputPath.toString(), Reporter.NULL);for (Record record : recordList) {//将Record记录组装成HDFS的TEXT行记录，列分隔符可自定义Text recordResult = new Text(StringUtils.join(mergeColumn(record), fieldDelimiter));writer.write(NullWritable.get(), recordResult);}writer.close(Reporter.NULL);
}
......
//基于内存的Channel实现
class  MemoryChannel{
......//HDFS写入端从Channel中Pull记录public void pullAll(Collection<Record> rs) {assert rs != null;rs.clear();try {lock.lockInterruptibly();while (this.queue.drainTo(rs, bufferSize) <= 0) {notEmpty.await(200L, TimeUnit.MILLISECONDS);}notInsufficient.signalAll();} catch (InterruptedException e) {throw new RuntimeException("pullAll", e);} finally {lock.unlock();}
}
......
}

上述方式是实现多文件并行传输的基础，每个独立Channel的传输过程互不影响，即使当前Chanel过程失败，也可以独立重跑恢复。

END


http://www.ppmy.cn/news/546764.html

相关文章

完美解码播放器（PotPlayer）关闭自动播放下一首的方法（找了好久）

完美解码播放器（PotPlayer）关闭自动播放下一首的方法（找了好久）

阅读更多...
完美解码终于出新版本了

完美解码终于出新版本了

新版本叫做完美者解码。。名字好山寨。一步到位的播放软件。
阅读更多...
固定完美解码播放器尺寸

固定完美解码播放器尺寸

右键“播放 ”“播放设置”“播放窗口尺寸 自己调或者选”勾选“仅在播放视频时调整”
阅读更多...
【经验分享】为什么视频画面解码失败之后显示的是绿幕？

【经验分享】为什么视频画面解码失败之后显示的是绿幕？

项目场景&#xff1a; 项目场景&#xff1a;我们在处理视频数据解码时&#xff0c;经常会遇到解码失败出现绿幕的情况&#xff0c;这个时候一般我们会去检查解码端的程序代码。 问题描述 在出现问题的时候&#xff0c;是否有考虑过以下的问题&#xff1a; “为什么是绿幕&a…
阅读更多...
php让视频自动全屏播放,完美解码怎么设置打开视频文件就全屏

php让视频自动全屏播放,完美解码怎么设置打开视频文件就全屏

想不想在电脑上实现打开视频文件就直接全屏播放&#xff0c;一般播放器做不到吧&#xff0c;完美解码这个万能的视频播放器就能实现&#xff0c;下面就教你如何设置。 软件名称&#xff1a;完美解码(PureCodec) 全能型影音解码包20200922软件大小&#xff1a;104MB更新时间&…
阅读更多...
关于kgm文件的解码

关于kgm文件的解码

寻找可以完成 kgm to mp3的解码源代码文件 这里的 kgm 是酷狗的音乐格式&#xff1f;现在酷狗没有开放解密算法 kgm 网传这其实就是mp3&#xff0c;不过是被酷狗加密过的&#xff0c;加密算法未知。 网传可以调用酷狗自己的库进行转换 https://download.csdn.net/download/we…
阅读更多...
音视频编解码技术（二）：AAC 音频编码技术

音视频编解码技术（二）：AAC 音频编码技术

一、AAC编码概述 AAC是高级音频编码&#xff08;Advanced Audio Coding&#xff09;的缩写&#xff0c;出现于1997年&#xff0c;最初是基于MPEG-2的音频编码技术&#xff0c;目的是取代MP3格式。2000年&#xff0c;MPEG-4标准出台&#xff0c;AAC重新集成了其它技术包括SBR或…
阅读更多...
FPGA硬件png图片解码器，支持所有颜色类型解码，提供工程源码和技术支持

FPGA硬件png图片解码器，支持所有颜色类型解码，提供工程源码和技术支持

目录 1、前言2、我已有的视频图像编解码方案3、png图片解码理论4、png图片解码性能介绍5、png图片解码器时序介绍输入码流输出图像信息和像素 6、png图片解码器设计详解7、工程源码和仿真工程代码架构仿真流程 8、输出RGB图像验证9、福利&#xff1a;工程代码的获取 1、前言 p…
阅读更多...
最新文章

Copyright @ 2022~2023