Skip to content

Commit

Permalink
修正多线程下载的bug
Browse files Browse the repository at this point in the history
  • Loading branch information
@leiurayer
leiurayer committed Jan 6, 2023
1 parent d8b747a commit 7b45274
Show file tree
Hide file tree
Showing 4 changed files with 84 additions and 77 deletions.
125 changes: 62 additions & 63 deletions src/DownKyi.Core/Downloader/MultiThreadDownloader.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,8 @@ public class MultiThreadDownloader
private bool _rangeAllowed;
private readonly HttpWebRequest _request;
private Action<HttpWebRequest> _requestConfigure = req => req.UserAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.122 Safari/537.36";
#endregion

#endregion 属性

#region 公共属性

Expand Down Expand Up @@ -68,11 +69,13 @@ public long TotalBytesReceived
{
try
{
return PartialDownloaderList.Where(t => t != null).Sum(t => t.TotalBytesRead);
lock (this)
{
return PartialDownloaderList.Where(t => t != null).Sum(t => t.TotalBytesRead);
}
}
catch (Exception e)
catch
{
Logging.LogManager.Error(e);
return 0;
}
}
Expand All @@ -91,7 +94,16 @@ public long TotalBytesReceived
/// <summary>
/// 下载速度
/// </summary>
public float TotalSpeedInBytes => PartialDownloaderList.Sum(t => t.SpeedInBytes);
public float TotalSpeedInBytes
{
get
{
lock (this)
{
return PartialDownloaderList.Sum(t => t.SpeedInBytes);
}
}
}

/// <summary>
/// 下载块
Expand All @@ -103,7 +115,7 @@ public long TotalBytesReceived
/// </summary>
public string FilePath { get; set; }

#endregion
#endregion 公共属性

#region 变量

Expand All @@ -124,7 +136,7 @@ public long TotalBytesReceived

private readonly AsyncOperation _aop;

#endregion
#endregion 变量

#region 下载管理器

Expand Down Expand Up @@ -166,7 +178,7 @@ public MultiThreadDownloader(string sourceUrl, int numOfParts) : this(sourceUrl,
{
}

#endregion
#endregion 下载管理器

#region 事件

Expand All @@ -181,16 +193,16 @@ private void temp_DownloadPartCompleted(object sender, EventArgs e)
return;
}

PartialDownloaderList.Sort((x, y) => y.RemainingBytes - x.RemainingBytes);
int rem = PartialDownloaderList[0].RemainingBytes;
PartialDownloaderList.Sort((x, y) => (int)(y.RemainingBytes - x.RemainingBytes));
var rem = PartialDownloaderList[0].RemainingBytes;
if (rem < 50 * 1024)
{
WaitOrResumeAll(PartialDownloaderList, false);
return;
}

int from = PartialDownloaderList[0].CurrentPosition + rem / 2;
int to = PartialDownloaderList[0].To;
var from = PartialDownloaderList[0].CurrentPosition + rem / 2;
var to = PartialDownloaderList[0].To;
if (from > to)
{
WaitOrResumeAll(PartialDownloaderList, false);
Expand All @@ -202,16 +214,19 @@ private void temp_DownloadPartCompleted(object sender, EventArgs e)
var temp = new PartialDownloader(_url, TempFileDirectory, Guid.NewGuid().ToString(), from, to, true);
temp.DownloadPartCompleted += temp_DownloadPartCompleted;
temp.DownloadPartProgressChanged += temp_DownloadPartProgressChanged;
PartialDownloaderList.Add(temp);
lock (this)
{
PartialDownloaderList.Add(temp);
}
temp.Start(_requestConfigure);
}

void temp_DownloadPartProgressChanged(object sender, EventArgs e)
private void temp_DownloadPartProgressChanged(object sender, EventArgs e)
{
UpdateProgress();
}

void UpdateProgress()
private void UpdateProgress()
{
int pr = (int)(TotalBytesReceived * 1d / Size * 100);
if (TotalProgress != pr)
Expand All @@ -224,11 +239,11 @@ void UpdateProgress()
}
}

#endregion
#endregion 事件

#region 方法

void CreateFirstPartitions()
private void CreateFirstPartitions()
{
Size = GetContentLength(ref _rangeAllowed, ref _url);
int maximumPart = (int)(Size / (25 * 1024));
Expand All @@ -247,17 +262,21 @@ void CreateFirstPartitions()
var temp = CreateNew(i, NumberOfParts, Size);
temp.DownloadPartProgressChanged += temp_DownloadPartProgressChanged;
temp.DownloadPartCompleted += temp_DownloadPartCompleted;
PartialDownloaderList.Add(temp);
lock (this)
{
PartialDownloaderList.Add(temp);
}
temp.Start(_requestConfigure);
}
}

void MergeParts()
private void MergeParts()
{
var mergeOrderedList = PartialDownloaderList.OrderBy(x => x.From);
var dir = new FileInfo(FilePath).DirectoryName;
Directory.CreateDirectory(dir);


using (var fs = File.OpenWrite(FilePath))
{
long totalBytesWrite = 0;
Expand Down Expand Up @@ -298,12 +317,12 @@ void MergeParts()
}
}

PartialDownloader CreateNew(int order, int parts, long contentLength)
private PartialDownloader CreateNew(int order, int parts, long contentLength)
{
int division = (int)contentLength / parts;
int remaining = (int)contentLength % parts;
int start = division * order;
int end = start + division - 1;
var division = contentLength / parts;
var remaining = contentLength % parts;
var start = division * order;
var end = start + division - 1;
end += order == parts - 1 ? remaining : 0;
return new PartialDownloader(_url, TempFileDirectory, Guid.NewGuid().ToString("N"), start, end, true);
}
Expand All @@ -315,15 +334,15 @@ PartialDownloader CreateNew(int order, int parts, long contentLength)
/// <param name="wait"></param>
public static void WaitOrResumeAll(List<PartialDownloader> list, bool wait)
{
foreach (var item in list)
for (var index = 0; index < list.Count; index++)
{
if (wait)
{
item.Wait();
list[index].Wait();
}
else
{
item.ResumeAfterWait();
list[index].ResumeAfterWait();
}
}
}
Expand All @@ -349,23 +368,21 @@ public long GetContentLength(ref bool rangeAllowed, ref string redirectedUrl)
_request.ServicePoint.ConnectionLimit = 4;
_requestConfigure(_request);

long ctl;
using (var resp = _request.GetResponse() as HttpWebResponse)
{
redirectedUrl = resp.ResponseUri.OriginalString;
ctl = resp.ContentLength;
var ctl = resp.ContentLength;
rangeAllowed = resp.Headers.AllKeys.Select((v, i) => new
{
HeaderName = v,
HeaderValue = resp.Headers[i]
}).Any(k => k.HeaderName.ToLower().Contains("range") && k.HeaderValue.ToLower().Contains("byte"));
_request.Abort();
return ctl;
}

return ctl;
}

#endregion
#endregion 方法

#region 公共方法

Expand All @@ -374,9 +391,12 @@ public long GetContentLength(ref bool rangeAllowed, ref string redirectedUrl)
/// </summary>
public void Pause()
{
foreach (var t in PartialDownloaderList.Where(t => !t.Completed))
lock (this)
{
t.Stop();
foreach (var t in PartialDownloaderList.Where(t => !t.Completed))
{
t.Stop();
}
}

Thread.Sleep(200);
Expand All @@ -385,34 +405,10 @@ public void Pause()
/// <summary>
/// 开始下载
/// </summary>
public void StartAsync()
{
//Task th = new Task(CreateFirstPartitions);
//th.Start();
StartAsync(false);
}

/// <summary>
/// 开始下载
/// </summary>
/// <param name="isWait"></param>
public void StartAsync(bool isWait)
public void Start()
{
Task th = new Task(CreateFirstPartitions);
th.Start();

if (isWait)
{
th.Wait();
}
}

/// <summary>
/// 开始下载
/// </summary>
public void Start()
{
CreateFirstPartitions();
}

/// <summary>
Expand All @@ -425,8 +421,8 @@ public void Resume()
{
if (PartialDownloaderList[i].Stopped)
{
int from = PartialDownloaderList[i].CurrentPosition + 1;
int to = PartialDownloaderList[i].To;
var from = PartialDownloaderList[i].CurrentPosition + 1;
var to = PartialDownloaderList[i].To;
if (from > to)
{
continue;
Expand All @@ -435,13 +431,16 @@ public void Resume()
var temp = new PartialDownloader(_url, TempFileDirectory, Guid.NewGuid().ToString(), from, to, _rangeAllowed);
temp.DownloadPartProgressChanged += temp_DownloadPartProgressChanged;
temp.DownloadPartCompleted += temp_DownloadPartCompleted;
PartialDownloaderList.Add(temp);
lock (this)
{
PartialDownloaderList.Add(temp);
}
PartialDownloaderList[i].To = PartialDownloaderList[i].CurrentPosition;
temp.Start(_requestConfigure);
}
}
}

#endregion
#endregion 公共方法
}
}
23 changes: 12 additions & 11 deletions src/DownKyi.Core/Downloader/PartialDownloader.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,11 +29,11 @@ public class PartialDownloader
public event EventHandler DownloadPartStopped;

private readonly AsyncOperation _aop = AsyncOperationManager.CreateOperation(null);
readonly int[] _lastSpeeds;
int _counter;
private readonly int[] _lastSpeeds;
private long _counter;
private long _to;
private long _totalBytesRead;
private bool _wait;
private int _to;
private int _totalBytesRead;

/// <summary>
/// 下载已停止
Expand Down Expand Up @@ -83,7 +83,7 @@ public class PartialDownloader
/// <summary>
/// to
/// </summary>
public int To
public long To
{
get => _to;
set
Expand All @@ -96,17 +96,17 @@ public int To
/// <summary>
/// from
/// </summary>
public int From { get; }
public long From { get; }

/// <summary>
/// 当前位置
/// </summary>
public int CurrentPosition => From + _totalBytesRead - 1;
public long CurrentPosition => From + _totalBytesRead - 1;

/// <summary>
/// 剩余字节数
/// </summary>
public int RemainingBytes => (int)(ContentLength - _totalBytesRead);
public long RemainingBytes => ContentLength - _totalBytesRead;

/// <summary>
/// 完整路径
Expand Down Expand Up @@ -139,7 +139,7 @@ public int SpeedInBytes
/// <param name="from"></param>
/// <param name="to"></param>
/// <param name="rangeAllowed"></param>
public PartialDownloader(string url, string dir, string fileGuid, int from, int to, bool rangeAllowed)
public PartialDownloader(string url, string dir, string fileGuid, long from, long to, bool rangeAllowed)
{
From = from;
_to = to;
Expand All @@ -150,7 +150,7 @@ public PartialDownloader(string url, string dir, string fileGuid, int from, int
_lastSpeeds = new int[10];
}

void DownloadProcedure(Action<HttpWebRequest> config)
private void DownloadProcedure(Action<HttpWebRequest> config)
{
using (var file = new FileStream(FullPath, FileMode.Create, FileAccess.ReadWrite, FileShare.Delete))
{
Expand All @@ -163,6 +163,7 @@ void DownloadProcedure(Action<HttpWebRequest> config)
req.ServicePoint.ConnectionLimit += 1;
req.ServicePoint.Expect100Continue = true;
req.ProtocolVersion = HttpVersion.Version11;
req.Proxy = WebRequest.GetSystemWebProxy();
config(req);
if (RangeAllowed)
{
Expand Down Expand Up @@ -265,4 +266,4 @@ public void ResumeAfterWait()
_wait = false;
}
}
}
}
Loading

0 comments on commit 7b45274

Please sign in to comment.