C# socket 多线程多管道可断点传送大文件(附单线程单管道传送)

这里只发布核心代码。源码及测试程序请点这里下载，谢谢。

有啥BUG，问题请发送email至ilovehaley.kid@gmail.com ，谢谢。:D

代码

#define Sleep
//#undef Sleep
//#define TransmitLog
#undef TransmitLog
//#define BreakpointLog
#undef BreakpointLog
using System;
using System.Net;
using System.Net.Sockets;
using System.IO;
using System.Text;
using System.Threading;
using System.Collections.Generic;
using System.Diagnostics;

namespace Rocky
{
public static class FileTransmiter
{
#region NestedType
private class SendWorker : IWorker
{
private long totalSent, totalSend;
private byte[] buffer;
private Socket sock;
private FileStream reader;
private Thread thread;
private bool isFinished;

public long TotalSent
{
get { return totalSent; }
}
public long TotalSend
{
get { return totalSend; }
}
public byte[] Buffer
{
get { return buffer; }
}
public Socket Client
{
get { return sock; }
}
public bool IsFinished
{
get { return isFinished; }
}

public SendWorker(IPEndPoint ip)
{
sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
sock.Connect(ip);
buffer = new byte[BufferSize];
}
public void Initialize(string path, long position, long length)
{
Initialize(path, position, length, 0L, length);
}
public void Initialize(string path, long position, long length, long worked, long total)
{
reader = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read);
reader.Position = position + worked;
totalSent = worked;
totalSend = total;
thread = new Thread(new ParameterizedThreadStart(Work));
thread.IsBackground = true;
#if TransmitLog
thread.Name = position.ToString() + length.ToString();
AppendTransmitLog(LogType.Transmit, thread.Name + " Initialized:" + totalSent + "/" + totalSend + ".");
#endif
}
private void Work(object obj)
{
int read, sent;
bool flag;
while (totalSent < totalSend)
{
read = reader.Read(buffer, 0, Math.Min(BufferSize, (int)(totalSend - totalSent)));
sent = 0;
flag = true;
while ((sent += sock.Send(buffer, sent, read, SocketFlags.None)) < read)
{
flag = false;
totalSent += (long)sent;
#if TransmitLog
AppendTransmitLog(LogType.Transmit, thread.Name + ":" + totalSent + "/" + totalSend + ".");
#endif
#if Sleep
Thread.Sleep(200);
#endif
}
if (flag)
{
totalSent += (long)read;
#if TransmitLog
AppendTransmitLog(LogType.Transmit, thread.Name + ":" + totalSent + "/" + totalSend + ".");
#endif
#if Sleep
Thread.Sleep(200);
#endif
}
}
reader.Dispose();
sock.Shutdown(SocketShutdown.Both);
sock.Close();
EventWaitHandle waitHandle = obj as EventWaitHandle;
if (waitHandle != null)
{
waitHandle.Set();
}
isFinished = true;
}

public void ReportProgress(out long worked, out long total)
{
worked = totalSent;
total = totalSend;
}

public void RunWork(EventWaitHandle waitHandle)
{
thread.Start(waitHandle);
}
}

private class ReceiveWorker : IWorker
{
private long offset, totalReceived, totalReceive;
private byte[] buffer;
private Socket sock;
private FileStream writer;
private Thread thread;
private bool isFinished;

public long TotalReceived
{
get { return totalReceived; }
}
public long TotalReceive
{
get { return totalReceive; }
}
public byte[] Buffer
{
get { return buffer; }
}
public Socket Client
{
get { return sock; }
}
public bool IsFinished
{
get { return isFinished; }
}

public ReceiveWorker(Socket client)
{
sock = client;
buffer = new byte[BufferSize];
}
public void Initialize(string path, long position, long length)
{
Initialize(path, position, length, 0L, length);
}
public void Initialize(string path, long position, long length, long worked, long total)
{
writer = new FileStream(path, FileMode.Open, FileAccess.Write, FileShare.Write);
writer.Position = position + worked;
writer.Lock(position, length);
offset = position;
totalReceived = worked;
totalReceive = total;
thread = new Thread(new ParameterizedThreadStart(Work));
thread.IsBackground = true;
#if TransmitLog
thread.Name = position.ToString() + length.ToString();
AppendTransmitLog(LogType.Transmit, thread.Name + " Initialized:" + totalReceived + "/" + totalReceive + ".");
#endif
}
private void Work(object obj)
{
int received;
while (totalReceived < totalReceive)
{
if ((received = sock.Receive(buffer)) == 0)
{
break;
}
writer.Write(buffer, 0, received);
writer.Flush();
totalReceived += (long)received;
#if TransmitLog
AppendTransmitLog(LogType.Transmit, thread.Name + ":" + totalReceived + "/" + totalReceive + ".");
#endif
#if Sleep
Thread.Sleep(200);
#endif
}
writer.Unlock(offset, totalReceive);
writer.Dispose();
sock.Shutdown(SocketShutdown.Both);
sock.Close();
EventWaitHandle waitHandle = obj as EventWaitHandle;
if (waitHandle != null)
{
waitHandle.Set();
}
isFinished = true;
}

public void ReportProgress(out long worked, out long total)
{
worked = totalReceived;
total = totalReceive;
}

public void RunWork(EventWaitHandle waitHandle)
{
thread.Start(waitHandle);
}
}

private interface IWorker
{
bool IsFinished { get; }
void Initialize(string path, long position, long length);
void Initialize(string path, long position, long length, long worked, long total);
void ReportProgress(out long worked, out long total);
void RunWork(EventWaitHandle waitHandle);
}
#endregion

#region Field
public const int BufferSize = 1024;
public const int PerLongCount = sizeof(long);
public const int MinThreadCount = 1;
public const int MaxThreadCount = 9;
public const string PointExtension = ".dat";
public const string TempExtension = ".temp";
private const long SplitSize = 1024L * 1024L * 100L;
public static readonly IPEndPoint TestIP;
#if TransmitLog
private static StreamWriter transmitLoger;
#endif
#if BreakpointLog
private static StreamWriter breakpointLoger;
#endif
#endregion

#region Constructor
static FileTransmiter()
{
AppDomain.CurrentDomain.UnhandledException += new UnhandledExceptionEventHandler(CurrentDomain_UnhandledException);
TestIP = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 520);
#if TransmitLog
transmitLoger = new StreamWriter(Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "transmit.log"), true, Encoding.Default);
#endif
#if BreakpointLog
breakpointLoger = new StreamWriter(Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "breakpoint.log"), true, Encoding.Default);
#endif
}

static void CurrentDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e)
{
StreamWriter writer = new StreamWriter(Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "exec.log"), true, Encoding.Default);
writer.Write("Time:");
writer.Write(DateTime.Now.ToShortTimeString());
writer.Write(". ");
writer.WriteLine(e.ExceptionObject);
writer.Dispose();
}

#region Log
#if TransmitLog || BreakpointLog
public enum LogType
{
Transmit,
Breakpoint
}

public static void AppendTransmitLog(LogType type, string msg)
{
switch (type)
{
case LogType.Transmit:
#if TransmitLog
transmitLoger.Write(DateTime.Now.ToShortTimeString());
transmitLoger.Write('\t');
transmitLoger.WriteLine(msg);
transmitLoger.Flush();
#endif
break;
case LogType.Breakpoint:
#if BreakpointLog
breakpointLoger.Write(DateTime.Now.ToShortTimeString());
breakpointLoger.Write('\t');
breakpointLoger.WriteLine(msg);
breakpointLoger.Flush();
#endif
break;
}
}
#endif
#endregion
#endregion

#region Single
public static void Send(IPEndPoint ip, string path)
{
Stopwatch watcher = new Stopwatch();
watcher.Start();
Socket sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
sock.Connect(ip);
byte[] buffer = new byte[BufferSize];
using (FileStream reader = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.None))
{
long send, length = reader.Length;
Buffer.BlockCopy(BitConverter.GetBytes(length), 0, buffer, 0, PerLongCount);
string fileName = Path.GetFileName(path);
sock.Send(buffer, 0, PerLongCount + Encoding.Default.GetBytes(fileName, 0, fileName.Length, buffer, PerLongCount), SocketFlags.None);
Console.WriteLine("Sending file:" + fileName + ".Plz wait...");
sock.Receive(buffer);
reader.Position = send = BitConverter.ToInt64(buffer, 0);
#if BreakpointLog
Console.WriteLine("Breakpoint " + reader.Position);
#endif
int read, sent;
bool flag;
while ((read = reader.Read(buffer, 0, BufferSize)) != 0)
{
sent = 0;
flag = true;
while ((sent += sock.Send(buffer, sent, read, SocketFlags.None)) < read)
{
flag = false;
send += (long)sent;
#if TransmitLog
Console.WriteLine("Sent " + send + "/" + length + ".");
#endif
#if Sleep
Thread.Sleep(200);
#endif
}
if (flag)
{
send += (long)read;
#if TransmitLog
Console.WriteLine("Sent " + send + "/" + length + ".");
#endif
#if Sleep
Thread.Sleep(200);
#endif
}
}
}
sock.Shutdown(SocketShutdown.Both);
sock.Close();
watcher.Stop();
Console.WriteLine("Send finish.Span Time:" + watcher.Elapsed.TotalMilliseconds + " ms.");
}

public static void Receive(IPEndPoint ip, string path)
{
Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
listener.Bind(ip);
listener.Listen(MinThreadCount);
Socket client = listener.Accept();
Stopwatch watcher = new Stopwatch();
watcher.Start();
byte[] buffer = new byte[BufferSize];
int received = client.Receive(buffer);
long receive, length = BitConverter.ToInt64(buffer, 0);
string fileName = Encoding.Default.GetString(buffer, PerLongCount, received - PerLongCount);
Console.WriteLine("Receiveing file:" + fileName + ".Plz wait...");
FileInfo file = new FileInfo(Path.Combine(path, fileName));
using (FileStream writer = file.Open(file.Exists ? FileMode.Append : FileMode.CreateNew, FileAccess.Write, FileShare.None))
{
receive = writer.Length;
client.Send(BitConverter.GetBytes(receive));
#if BreakpointLog
Console.WriteLine("Breakpoint " + receive);
#endif
while (receive < length)
{
if ((received = client.Receive(buffer)) == 0)
{
Console.WriteLine("Send Stop.");
return;
}
writer.Write(buffer, 0, received);
writer.Flush();
receive += (long)received;
#if TransmitLog
Console.WriteLine("Received " + receive + "/" + length + ".");
#endif
#if Sleep
Thread.Sleep(200);
#endif
}
}
client.Shutdown(SocketShutdown.Both);
client.Close();
watcher.Stop();
Console.WriteLine("Receive finish.Span Time:" + watcher.Elapsed.TotalMilliseconds + " ms.");
}
#endregion

#region Supper
#region Extensions
private static int ReportProgress(this IWorker[] workers, out long worked, out long total)
{
worked = total = 0L;
long w, t;
foreach (IWorker worker in workers)
{
worker.ReportProgress(out w, out t);
worked += w;
total += t;
}
return (int)(worked / total) * 100;
}
private static int ReportSpeed(this IWorker[] workers, ref long oldValue)
{
long w, t;
workers.ReportProgress(out w, out t);
int speed = (int)((w - oldValue) / 8L);
oldValue = w;
return speed;
}
private static bool IsAllFinished(this IWorker[] workers)
{
bool flag = true;
foreach (IWorker worker in workers)
{
if (!worker.IsFinished)
{
flag = false;
break;
}
}
return flag;
}
#endregion

#region Helper
public static void Write(long value, byte[] buffer, int offset)
{
buffer[offset++] = (byte)value;
buffer[offset++] = (byte)(value >> 8);
buffer[offset++] = (byte)(value >> 0x10);
buffer[offset++] = (byte)(value >> 0x18);
buffer[offset++] = (byte)(value >> 0x20);
buffer[offset++] = (byte)(value >> 40);
buffer[offset++] = (byte)(value >> 0x30);
buffer[offset] = (byte)(value >> 0x38);
}
public static void Read(out long value, byte[] buffer, int offset)
{
uint num = (uint)(((buffer[offset++] | (buffer[offset++] << 8)) | (buffer[offset++] << 0x10)) | (buffer[offset++] << 0x18));
uint num2 = (uint)(((buffer[offset++] | (buffer[offset++] << 8)) | (buffer[offset++] << 0x10)) | (buffer[offset] << 0x18));
value = (long)((num2 << 0x20) | num);
}
#endregion

public static int GetThreadCount(long fileSize)
{
int count = (int)(fileSize / SplitSize);
if (count < MinThreadCount)
{
count = MinThreadCount;
}
else if (count > MaxThreadCount)
{
count = MaxThreadCount;
}
return count;
}

public static void SupperSend(IPEndPoint ip, string path)
{
Stopwatch watcher = new Stopwatch();
watcher.Start();
FileInfo file = new FileInfo(path);
#if DEBUG
if (!file.Exists)
{
throw new FileNotFoundException();
}
#endif
SendWorker worker = new SendWorker(ip);
long fileLength = file.Length;
Buffer.BlockCopy(BitConverter.GetBytes(fileLength), 0, worker.Buffer, 0, PerLongCount);
string fileName = file.Name;
worker.Client.Send(worker.Buffer, 0, PerLongCount + Encoding.Default.GetBytes(fileName, 0, fileName.Length, worker.Buffer, PerLongCount), SocketFlags.None);
Console.WriteLine("Sending file:" + fileName + ".Plz wait...");
int threadCount = GetThreadCount(fileLength);
SendWorker[] workers = new SendWorker[threadCount];
for (int i = 0; i < threadCount; i++)
{
workers[i] = i == 0 ? worker : new SendWorker(ip);
}
#region Breakpoint
int perPairCount = PerLongCount * 2, count = perPairCount * threadCount;
byte[] bufferInfo = new byte[count];
long oddSize, avgSize = Math.DivRem(fileLength, (long)threadCount, out oddSize);
if (worker.Client.Receive(bufferInfo) == 4)
{
for (int i = 0; i < threadCount; i++)
{
workers[i].Initialize(path, i * avgSize, i == threadCount - 1 ? avgSize + oddSize : avgSize);
}
}
else
{
long w, t;
for (int i = 0; i < threadCount; i++)
{
Read(out w, bufferInfo, i * perPairCount);
Read(out t, bufferInfo, i * perPairCount + PerLongCount);
workers[i].Initialize(path, i * avgSize, i == threadCount - 1 ? avgSize + oddSize : avgSize, w, t);
#if BreakpointLog
AppendTransmitLog(LogType.Breakpoint, i + " read:" + w + "/" + t + ".");
#endif
}
}
Thread.Sleep(200);
#endregion
AutoResetEvent reset = new AutoResetEvent(true);
for (int i = 0; i < threadCount; i++)
{
workers[i].RunWork(i == threadCount - 1 ? reset : null);
}
reset.WaitOne();
#region Breakpoint
int speed;
long value = 0L;
do
{
speed = workers.ReportSpeed(ref value);
Console.WriteLine("waiting for other threads. Progress:" + value + "/" + fileLength + ";Speed:" + speed + "kb/s.");
Thread.Sleep(1000);
}
while (!workers.IsAllFinished());
speed = workers.ReportSpeed(ref value);
Console.WriteLine("waiting for other threads. Progress:" + value + "/" + fileLength + ";Speed:" + speed + "kb/s.");
#endregion
watcher.Stop();
Console.WriteLine("Send finish.Span Time:" + watcher.Elapsed.TotalMilliseconds + " ms.");
}

public static void SupperReceive(IPEndPoint ip, string path)
{
Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
listener.Bind(ip);
listener.Listen(MaxThreadCount);
ReceiveWorker worker = new ReceiveWorker(listener.Accept());
Stopwatch watcher = new Stopwatch();
watcher.Start();
int recv = worker.Client.Receive(worker.Buffer);
long fileLength = BitConverter.ToInt64(worker.Buffer, 0);
string fileName = Encoding.Default.GetString(worker.Buffer, PerLongCount, recv - PerLongCount);
Console.WriteLine("Receiveing file:" + fileName + ".Plz wait...");
int threadCount = GetThreadCount(fileLength);
ReceiveWorker[] workers = new ReceiveWorker[threadCount];
for (int i = 0; i < threadCount; i++)
{
workers[i] = i == 0 ? worker : new ReceiveWorker(listener.Accept());
}
#region Breakpoint
int perPairCount = PerLongCount * 2, count = perPairCount * threadCount;
byte[] bufferInfo = new byte[count];
string filePath = Path.Combine(path, fileName), pointFilePath = Path.ChangeExtension(filePath, PointExtension), tempFilePath = Path.ChangeExtension(filePath, TempExtension);
FileStream pointStream;
long oddSize, avgSize = Math.DivRem(fileLength, (long)threadCount, 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 w, t;
for (int i = 0; i < threadCount; i++)
{
Read(out w, bufferInfo, i * perPairCount);
Read(out t, bufferInfo, i * perPairCount + PerLongCount);
workers[i].Initialize(tempFilePath, i * avgSize, i == threadCount - 1 ? avgSize + oddSize : avgSize, w, t);
#if BreakpointLog
AppendTransmitLog(LogType.Breakpoint, i + " read:" + w + "/" + t + ".");
#endif
}
worker.Client.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(fileLength);
stream.Flush();
stream.Dispose();
for (int i = 0; i < threadCount; i++)
{
workers[i].Initialize(tempFilePath, i * avgSize, i == threadCount - 1 ? avgSize + oddSize : avgSize);
}
worker.Client.Send(bufferInfo, 0, 4, SocketFlags.None);
}
Timer timer = new Timer(state =>
{
long w, t;
for (int i = 0; i < threadCount; i++)
{
workers[i].ReportProgress(out w, out t);
Write(w, bufferInfo, i * perPairCount);
Write(t, bufferInfo, i * perPairCount + PerLongCount);
#if BreakpointLog
AppendTransmitLog(LogType.Breakpoint, i + " write:" + w + "/" + t + ".");
#endif
}
pointStream.Position = 0L;
pointStream.Write(bufferInfo, 0, count);
pointStream.Flush();

}, null, TimeSpan.Zero, TimeSpan.FromSeconds(2));
#endregion
AutoResetEvent reset = new AutoResetEvent(true);
for (int i = 0; i < threadCount; i++)
{
workers[i].RunWork(i == threadCount - 1 ? reset : null);
}
reset.WaitOne();
#region Breakpoint
int speed;
long value = 0L;
do
{
speed = workers.ReportSpeed(ref value);
Console.WriteLine("waiting for other threads. Progress:" + value + "/" + fileLength + ";Speed:" + speed + "kb/s.");
Thread.Sleep(1000);
}
while (!workers.IsAllFinished());
speed = workers.ReportSpeed(ref value);
Console.WriteLine("waiting for other threads. Progress:" + value + "/" + fileLength + ";Speed:" + speed + "kb/s.");
timer.Dispose();
pointStream.Dispose();
File.Delete(pointFilePath);
File.Move(tempFilePath, filePath);
#endregion
watcher.Stop();
Console.WriteLine("Receive finish.Span Time:" + watcher.Elapsed.TotalMilliseconds + " ms.");
}
#endregion
}
}

PS：
1、通过测试发现多线程stream.Flush()的地方会阻塞，可以尝试增大stream的缓冲区或手动管理调用stream的Flush()。
2、断点测试方法：先打开Server端，然后打开Client端发送；中途任意关闭一端，再重复前一步即可。