基于C#实现的多线程文件上传下载工具

<!-- fluentftp 用于ftp操作 -->
<packagereference include="fluentftp" version="39.0.0" />
<!-- newtonsoft.json 用于配置管理 -->
<packagereference include="newtonsoft.json" version="13.0.3" />

using fluentftp;
using system;
using system.io;
using system.threading.tasks;

public class ftpservice : idisposable
{
    private readonly ftpclient _client;
    private readonly semaphoreslim _semaphore;

    public ftpservice(string host, string user, string pass, int maxconcurrent = 5)
    {
        _client = new ftpclient(host, user, pass);
        _semaphore = new semaphoreslim(maxconcurrent);
    }

    public async task<bool> uploadfile(string localpath, string remotepath, bool resume = false)
    {
        await _semaphore.waitasync();
        try
        {
            using var filestream = file.openread(localpath);
            var remotesize = await _client.getfilesizeasync(remotepath);
            
            if (resume && remotesize > 0)
            {
                filestream.seek(remotesize, seekorigin.begin);
                await _client.uploadfileasync(filestream, remotepath, ftpremoteexists.append, true);
            }
            else
            {
                await _client.uploadfileasync(filestream, remotepath, ftpremoteexists.overwrite, true);
            }
            return true;
        }
        catch (exception ex)
        {
            console.writeline($"上传失败: {ex.message}");
            return false;
        }
        finally
        {
            _semaphore.release();
        }
    }

    public async task<bool> downloadfile(string remotepath, string localpath, bool resume = false)
    {
        await _semaphore.waitasync();
        try
        {
            using var filestream = file.openwrite(localpath);
            long remotesize = await _client.getfilesizeasync(remotepath);
            long localsize = file.exists(localpath) ? new fileinfo(localpath).length : 0;

            if (resume && localsize < remotesize)
            {
                await _client.downloadfileasync(filestream, remotepath, ftplocalexists.append, ftpverify.retry);
            }
            else
            {
                await _client.downloadfileasync(filestream, remotepath, ftplocalexists.create, ftpverify.retry);
            }
            return true;
        }
        catch (exception ex)
        {
            console.writeline($"下载失败: {ex.message}");
            return false;
        }
        finally
        {
            _semaphore.release();
        }
    }

    public void dispose()
    {
        _client?.dispose();
        _semaphore?.dispose();
    }
}

using system.collections.concurrent;
using system.threading;
using system.threading.tasks;

public class transfermanager : idisposable
{
    private readonly concurrentqueue<transfertask> _uploadqueue = new();
    private readonly concurrentqueue<transfertask> _downloadqueue = new();
    private readonly cancellationtokensource _cts = new();
    private readonly ftpservice _ftpservice;

    public transfermanager(ftpservice ftpservice)
    {
        _ftpservice = ftpservice;
        startprocessing();
    }

    public void enqueueupload(string localpath, string remotepath)
    {
        _uploadqueue.enqueue(new transfertask(localpath, remotepath, transfertype.upload));
    }

    public void enqueuedownload(string remotepath, string localpath)
    {
        _downloadqueue.enqueue(new transfertask(remotepath, localpath, transfertype.download));
    }

    private async void startprocessing()
    {
        while (!_cts.token.iscancellationrequested)
        {
            if (_uploadqueue.trydequeue(out var task))
            {
                await processtask(task);
            }
            else if (_downloadqueue.trydequeue(out task))
            {
                await processtask(task);
            }
            await task.delay(100);
        }
    }

    private async task processtask(transfertask task)
    {
        try
        {
            bool success = task.type == transfertype.upload
                ? await _ftpservice.uploadfile(task.source, task.destination)
                : await _ftpservice.downloadfile(task.source, task.destination);

            ontaskcompleted?.invoke(task, success);
        }
        catch
        {
            ontaskfailed?.invoke(task);
        }
    }

    public event action<transfertask, bool> ontaskcompleted;
    public event action<transfertask> ontaskfailed;

    public void dispose()
    {
        _cts.cancel();
        _ftpservice?.dispose();
    }
}

public enum transfertype { upload, download }
public class transfertask
{
    public string source { get; }
    public string destination { get; }
    public transfertype type { get; }

    public transfertask(string source, string destination, transfertype type)
    {
        source = source;
        destination = destination;
        type = type;
    }
}

<window x:class="filetransfertool.mainwindow"
        title="多线程文件传输工具" height="450" width="800">
    <dockpanel>
        <statusbar dockpanel.dock="bottom">
            <textblock text="{binding statustext}"/>
            <progressbar value="{binding progress}" width="200"/>
        </statusbar>
        
        <tabcontrol>
            <tabitem header="上传队列">
                <datagrid itemssource="{binding uploadtasks}" autogeneratecolumns="false">
                    <datagrid.columns>
                        <datagridtextcolumn header="文件名" binding="{binding source}"/>
                        <datagridtextcolumn header="进度" binding="{binding progress}"/>
                    </datagrid.columns>
                </datagrid>
            </tabitem>
            <tabitem header="下载队列">
                <!-- 下载队列界面 -->
            </tabitem>
        </tabcontrol>
    </dockpanel>
</window>

public class transferviewmodel : inotifypropertychanged
{
    private int _progress;
    public int progress
    {
        get => _progress;
        set
        {
            _progress = value;
            onpropertychanged(nameof(progress));
        }
    }

    private string _statustext = "就绪";
    public string statustext
    {
        get => _statustext;
        set
        {
            _statustext = value;
            onpropertychanged(nameof(statustext));
        }
    }

    public observablecollection<transfertask> uploadtasks { get; } = new();
    public observablecollection<transfertask> downloadtasks { get; } = new();

    public event propertychangedeventhandler propertychanged;
    protected virtual void onpropertychanged(string propertyname)
    {
        propertychanged?.invoke(this, new propertychangedeventargs(propertyname));
    }
}

{
  "ftpsettings": {
    "host": "ftp.example.com",
    "username": "user",
    "password": "pass",
    "maxconcurrent": 8,
    "buffersize": 8388608
  },
  "transferpaths": {
    "uploaddir": "d:\\uploads",
    "downloaddir": "d:\\downloads"
  }
}

filetransfertool.exe --mode server --port 2121 --root c:\shared
filetransfertool.exe --mode client --host ftp.example.com --user user --pass pass