更新使用.net 4.0 Parallel 来代替 new Thread()；
/********************************************************************************* ** File Name : FileTransmitor.cs ** Copyright (C) 2010 Snda Network Corporation. All Rights Reserved. ** Creator : RockyWong ** Create Date : 2010-06-02 11:22:45 ** Update Date : 2013-01-11 11:35:26 ** Description : 多线程多管道可断点传输大文件 ** Version No : *********************************************************************************/ using System; using System.Collections.Generic; using System.Diagnostics.Contracts; using System.IO; using System.Linq; using System.Net; using System.Net.Sockets; using System.Text; using System.Threading; using System.Threading.Tasks; namespace Rocky.Net { public sealed class FileTransfer : Disposable { #region Fields internal const int PerLongSize = sizeof(long); internal const string PointExtension = ".dat"; internal const string TempExtension = ".temp"; private string _savePath; private Socket _listener; #endregion #region Properties public event EventHandler<TransferEventArgs> Prepare; public event EventHandler<TransferEventArgs> ProgressChanged; public event EventHandler<TransferEventArgs> Completed; public string DirectoryPath { get { return _savePath; } set { _savePath = value + @"\" + DateTime.Now.ToString("yyyy-MM") + @"\"; } } #endregion #region Constructors public FileTransfer() { } protected override void DisposeInternal(bool disposing) { if (disposing) { SocketHelper.DisposeListener(ref _listener); } _listener = null; Prepare = null; ProgressChanged = null; Completed = null; } #endregion #region Methods private void OnPrepare(TransferEventArgs e) { if (this.Prepare != null) { this.Prepare(this, e); } } private void OnProgressChanged(TransferEventArgs e) { if (this.ProgressChanged != null) { this.ProgressChanged(this, e); } } private void OnCompleted(TransferEventArgs e) { if (this.Completed != null) { this.Completed(this, e); } } #endregion #region Receive /// <summary> /// Listen & Receive /// </summary> /// <param name="savePath"></param> /// <param name="port"></param> public void Listen(string savePath, ushort port) { Contract.Requires(!string.IsNullOrEmpty(savePath)); if (_listener != null) { throw new ApplicationException("已启动监听"); } Runtime.CreateDirectory(_savePath = savePath); var localIpe = new IPEndPoint(IPAddress.Any, port); //最多支持16线程 _listener = SocketHelper.CreateListener(localIpe, 16); TaskHelper.Factory.StartNew(() => { while (_listener != null) { Socket controlClient = _listener.Accept(); Runtime.LogInfo("TunnelTest 双工通讯: {0}.", controlClient.RemoteEndPoint); TransferConfig config; controlClient.Receive(out config); var e = new TransferEventArgs(config); this.OnPrepare(e); if (e.Cancel) { controlClient.Close(); continue; } var chunkGroup = new ReceiveChunkModel[config.ThreadCount]; chunkGroup[0] = new ReceiveChunkModel(controlClient); for (int i = 1; i < chunkGroup.Length; i++) { chunkGroup[i] = new ReceiveChunkModel(_listener.Accept()); } TaskHelper.Factory.StartNew(Receive, new object[] { e, chunkGroup }); } }); } private void Receive(object state) { var args = (object[])state; var e = (TransferEventArgs)args[0]; var chunkGroup = (ReceiveChunkModel[])args[1]; var controlClient = chunkGroup[0].Client; e.Progress = new TransferProgress(); e.Progress.Start(e.Config.FileLength); #region Breakpoint int perPairCount = PerLongSize * 2, count = perPairCount * chunkGroup.Length; byte[] bufferInfo = new byte[count]; string filePath = Path.Combine(_savePath, e.Config.Checksum + Path.GetExtension(e.Config.FileName)), pointFilePath = Path.ChangeExtension(filePath, PointExtension), tempFilePath = Path.ChangeExtension(filePath, TempExtension); FileStream pointStream; long oddSize, avgSize = Math.DivRem(e.Config.FileLength, (long)chunkGroup.Length, out oddSize); if (File.Exists(pointFilePath) && File.Exists(tempFilePath)) { pointStream = new FileStream(pointFilePath, FileMode.Open, FileAccess.ReadWrite, FileShare.None); pointStream.Read(bufferInfo, 0, count); long fValue, tValue; for (int i = 0, j = chunkGroup.Length - 1; i < chunkGroup.Length; i++) { fValue = BitConverter.ToInt64(bufferInfo, i * perPairCount); tValue = BitConverter.ToInt64(bufferInfo, i * perPairCount + PerLongSize); chunkGroup[i].Initialize(tempFilePath, e.Config.ChunkLength, i * avgSize, i == j ? avgSize + oddSize : avgSize, fValue, tValue); Runtime.LogDebug("[Multi]Local{0} breakpoint read{1}:{2}/{3}.", _listener.LocalEndPoint, i, fValue, tValue); } controlClient.Send(bufferInfo); } else { pointStream = new FileStream(pointFilePath, FileMode.CreateNew, FileAccess.ReadWrite, FileShare.None); FileStream stream = new FileStream(tempFilePath, FileMode.CreateNew, FileAccess.Write, FileShare.Write); stream.SetLength(e.Config.FileLength); stream.Flush(); stream.Dispose(); for (int i = 0, j = chunkGroup.Length - 1; i < chunkGroup.Length; i++) { chunkGroup[i].Initialize(tempFilePath, e.Config.ChunkLength, i * avgSize, i == j ? avgSize + oddSize : avgSize); } controlClient.Send(bufferInfo, 0, 4, SocketFlags.None); } var timer = new Timer(arg => { long fValue, tValue; for (int i = 0; i < chunkGroup.Length; i++) { chunkGroup[i].ReportProgress(out fValue, out tValue); Buffer.BlockCopy(BitConverter.GetBytes(fValue), 0, bufferInfo, i * perPairCount, 8); Buffer.BlockCopy(BitConverter.GetBytes(tValue), 0, bufferInfo, i * perPairCount + PerLongSize, 8); Runtime.LogDebug("[Multi]Local{0} breakpoint write{1}:{2}/{3}.", _listener.LocalEndPoint, i, fValue, tValue); } pointStream.Position = 0L; pointStream.Write(bufferInfo, 0, count); pointStream.Flush(); }, null, TimeSpan.Zero, TimeSpan.FromSeconds(4)); #endregion Parallel.ForEach(chunkGroup, chunk => chunk.Run()); long bytesTransferred = 0L; do { chunkGroup.ReportSpeed(e.Progress, ref bytesTransferred); this.OnProgressChanged(e); Thread.Sleep(1000); } while (!chunkGroup.IsAllCompleted()); chunkGroup.ReportSpeed(e.Progress, ref bytesTransferred); this.OnProgressChanged(e); timer.Dispose(); pointStream.Dispose(); File.Delete(pointFilePath); File.Move(tempFilePath, filePath); e.Progress.Stop(); this.OnCompleted(e); } #endregion #region Send public void Send(TransferConfig config, IPEndPoint remoteIpe) { Contract.Requires(config != null && remoteIpe != null); var controlChunk = new SendChunkModel(remoteIpe); controlChunk.Client.Send(config); var e = new TransferEventArgs(config); this.OnPrepare(e); if (e.Cancel || !controlChunk.Client.Connected) { controlChunk.Client.Close(); return; } var chunkGroup = new SendChunkModel[config.ThreadCount]; chunkGroup[0] = controlChunk; for (int i = 1; i < chunkGroup.Length; i++) { chunkGroup[i] = new SendChunkModel(remoteIpe); } e.Progress = new TransferProgress(); e.Progress.Start(config.FileLength); #region Breakpoint int perPairCount = PerLongSize * 2, count = perPairCount * chunkGroup.Length; byte[] bufferInfo = new byte[count]; long oddSize, avgSize = Math.DivRem(config.FileLength, (long)chunkGroup.Length, out oddSize); if (controlChunk.Client.Receive(bufferInfo) == 4) { for (int i = 0, j = chunkGroup.Length - 1; i < chunkGroup.Length; i++) { chunkGroup[i].Initialize(config.FilePath, config.ChunkLength, i * avgSize, i == j ? avgSize + oddSize : avgSize); } } else { long fValue, tValue; for (int i = 0, j = chunkGroup.Length - 1; i < chunkGroup.Length; i++) { fValue = BitConverter.ToInt64(bufferInfo, i * perPairCount); tValue = BitConverter.ToInt64(bufferInfo, i * perPairCount + PerLongSize); chunkGroup[i].Initialize(config.FilePath, config.ChunkLength, i * avgSize, i == j ? avgSize + oddSize : avgSize, fValue, tValue); Runtime.LogDebug("[Multi]Remote{0} breakpoint{1}:{2}/{3}.", remoteIpe, i, fValue, tValue); } } Thread.Sleep(200); #endregion Parallel.ForEach(chunkGroup, chunk => chunk.Run()); long bytesTransferred = 0L; do { chunkGroup.ReportSpeed(e.Progress, ref bytesTransferred); this.OnProgressChanged(e); Thread.Sleep(1000); } while (!chunkGroup.IsAllCompleted()); chunkGroup.ReportSpeed(e.Progress, ref bytesTransferred); this.OnProgressChanged(e); e.Progress.Stop(); this.OnCompleted(e); } #endregion } }
另外推荐一个网络小工具：
内测安装地址：http://publish.xineworld.com/cloudagent/publish.htm
上篇文章：http://www.cnblogs.com/Googler/archive/2013/01/11/2856219.html