我需要从不同的线程中按缓冲区逐个向文件写入数据。为了避免锁定,我将写入到不同的文件中,比如说 'file_1','file_2',最后将它们合并成一个文件 'file'。这种方法是否可行?有更好的建议吗?
一些文件非常巨大,包含数千个缓冲区。因此会创建成千上万个临时文件,然后再合并和清理。
我需要从不同的线程中按缓冲区逐个向文件写入数据。为了避免锁定,我将写入到不同的文件中,比如说 'file_1','file_2',最后将它们合并成一个文件 'file'。这种方法是否可行?有更好的建议吗?
一些文件非常巨大,包含数千个缓冲区。因此会创建成千上万个临时文件,然后再合并和清理。
我的直觉是文件的处理会很耗时,而且管理成千上万个文件听起来很复杂容易出错。
那么,我们可以考虑让一个专门的线程进行写入操作。其他线程只需将它们的消息添加到等待写入的队列中即可。虽然需要一些同步开销,但锁中实际执行的工作非常小,只是将"指针"拷贝到消息队列中。打开文件并写入它们可能比获取互斥锁还要耗费更多时间,因此这种方法实际上可能会提高性能。
BlockingCollection
来管理一个缓冲区队列以写入文件。ParallelFileWriter
,然后在所有想要写入文件的线程中使用它。完成后,只需处置它(但请确保在所有线程都完成写入之前不要处置它!)。using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace Demo
{
public sealed class ParallelFileWriter: IDisposable
{
// maxQueueSize is the maximum number of buffers you want in the queue at once.
// If this value is reached, any threads calling Write() will block until there's
// room in the queue.
public ParallelFileWriter(string filename, int maxQueueSize)
{
_stream = new FileStream(filename, FileMode.Create);
_queue = new BlockingCollection<byte[]>(maxQueueSize);
_writerTask = Task.Run(() => writerTask());
}
public void Write(byte[] data)
{
_queue.Add(data);
}
public void Dispose()
{
_queue.CompleteAdding();
_writerTask.Wait();
_stream.Close();
}
private void writerTask()
{
foreach (var data in _queue.GetConsumingEnumerable())
{
Debug.WriteLine("Queue size = {0}", _queue.Count);
_stream.Write(data, 0, data.Length);
}
}
private readonly Task _writerTask;
private readonly BlockingCollection<byte[]> _queue;
private readonly FileStream _stream;
}
class Program
{
private void run()
{
// For demo purposes, cancel after a couple of seconds.
using (var fileWriter = new ParallelFileWriter(@"C:\TEST\TEST.DATA", 100))
using (var cancellationSource = new CancellationTokenSource(2000))
{
const int NUM_THREADS = 8;
Action[] actions = new Action[NUM_THREADS];
for (int i = 0; i < NUM_THREADS; ++i)
{
int id = i;
actions[i] = () => writer(cancellationSource.Token, fileWriter, id);
}
Parallel.Invoke(actions);
}
}
private void writer(CancellationToken cancellation, ParallelFileWriter fileWriter, int id)
{
int index = 0;
while (!cancellation.IsCancellationRequested)
{
string text = string.Format("{0}:{1}\n", id, index++);
byte[] data = Encoding.UTF8.GetBytes(text);
fileWriter.Write(data);
}
}
static void Main(string[] args)
{
new Program().run();
}
}
}