VGe*_*iev 18 delphi macos upload multithreading firemonkey
目前我正在使用Delphi XE3客户端/服务器应用程序来传输文件(使用Indy FTP组件).客户端部分监视文件夹,获取内部文件列表,将它们上载到服务器并删除原始文件.上传由一个单独的线程完成,该线程逐个处理文件.文件的范围从0到几千,它们的大小也有很大差异.
它是为OSX和Windows编译的Firemonkey应用程序,因此我不得不使用TThread而不是OmniThreadLibrary,这是我更喜欢的.我的客户报告该应用程序随机冻结.我无法复制它,但由于我对TThread没有那么多经验,我可能会把死锁条件放在某处.我读了很多例子,但我仍然不确定一些多线程的细节.
应用程序结构很简单:
主线程中的计时器检查文件夹并将每个文件的信息获取到记录中,该记录进入通用TList.此列表保留有关文件名称,大小,进度,文件是完全上载还是必须重试的信息.所有显示在带有进度条等的网格中.此列表仅由主线程访问.之后,通过调用AddFile方法(下面的代码)将列表中的项目发送到线程.该线程将所有文件存储在一个线程安全的队列中,如下所示http://delphihaven.wordpress.com/2011/05/06/using-tmonitor-2/
上传文件时,上传者线程通知主线程调用同步.
主线程定期调用Uploader.GetProgress方法来检查当前文件进度并显示它.这个函数实际上不是线程安全的,但它可能导致死锁,还是只返回错误的数据？
什么是安全有效的进度检查方法？
那么这种方法还可以,或者我错过了什么？你会怎么做？
例如,我虽然只是为了阅读文件夹内容而创建一个新线程.这意味着我使用的TList必须是线程安全的,但必须始终访问它以刷新GUI网格中显示的信息.不是所有的同步都会降低GUI的速度吗？
我已经发布了下面的简化代码,以防有人想看一下.如果没有,我会很高兴听到一些关于我应该使用的一般意见.主要目标是在OSX和Windows上工作; 能够显示有关所有文件的信息和当前文件的进度; 并且无论文件的数量和大小如何都要响应.
这是上传者线程的代码.我删除了一些以便于阅读:
type
TFileStatus = (fsToBeQueued, fsUploaded, fsQueued);
TFileInfo = record
ID: Integer;
Path: String;
Size: Int64;
UploadedSize: Int64;
Status: TFileStatus;
end;
TUploader = class(TThread)
private
FTP: TIdFTP;
fQueue: TThreadedQueue<TFileInfo>;
fCurrentFile: TFileInfo;
FUploading: Boolean;
procedure ConnectFTP;
function UploadFile(aFileInfo: TFileInfo): String;
procedure OnFTPWork(ASender: TObject; AWorkMode: TWorkMode; AWorkCount: Int64);
procedure SignalComplete;
procedure SignalError(aError: String);
protected
procedure Execute; override;
public
property Uploading: Boolean read FUploading;
constructor Create;
destructor Destroy; override;
procedure Terminate;
procedure AddFile(const aFileInfo: TFileInfo);
function GetProgress: TFileInfo;
end;
procedure TUploader.AddFile(const aFileInfo: TFileInfo);
begin
fQueue.Enqueue(aFileInfo);
end;
procedure TUploader.ConnectFTP;
begin
...
FTP.Connect;
end;
constructor TUploader.Create;
begin
inherited Create(false);
FreeOnTerminate := false;
fQueue := TThreadedQueue<TFileInfo>.Create;
// Create the TIdFTP and set ports and other params
...
end;
destructor TUploader.Destroy;
begin
fQueue.Close;
fQueue.Free;
FTP.Free;
inherited;
end;
// Process the whole queue and inform the main thread of the progress
procedure TUploader.Execute;
var
Temp: TFileInfo;
begin
try
ConnectFTP;
except
on E: Exception do
SignalError(E.Message);
end;
// Use Peek instead of Dequeue, because the item should not be removed from the queue if it fails
while fQueue.Peek(fCurrentFile) = wrSignaled do
try
if UploadFile(fCurrentFile) = '' then
begin
fQueue.Dequeue(Temp); // Delete the item from the queue if succesful
SignalComplete;
end;
except
on E: Exception do
SignalError(E.Message);
end;
end;
// Return the current file's info to the main thread. Used to update the progress indicators
function TUploader.GetProgress: TFileInfo;
begin
Result := fCurrentFile;
end;
// Update the uploaded size for the current file. This information is retrieved by a timer from the main thread to update the progress bar
procedure TUploader.OnFTPWork(ASender: TObject; AWorkMode: TWorkMode; AWorkCount: Int64);
begin
fCurrentFile.UploadedSize := AWorkCount;
end;
procedure TUploader.SignalComplete;
begin
Synchronize(
procedure
begin
frmClientMain.OnCompleteFile(fCurrentFile);
end);
end;
procedure TUploader.SignalError(aError: String);
begin
try
FTP.Disconnect;
except
end;
if fQueue.Closed then
Exit;
Synchronize(
procedure
begin
frmClientMain.OnUploadError(aError);
end);
end;
// Clear the queue and terminate the thread
procedure TUploader.Terminate;
begin
fQueue.Close;
inherited;
end;
function TUploader.UploadFile(aFileInfo: TFileInfo): String;
begin
Result := 'Error';
try
if not FTP.Connected then
ConnectFTP;
FUploading := true;
FTP.Put(aFileInfo.Path, ExtractFileName(aFileInfo.Path));
Result := '';
finally
FUploading := false;
end;
end;
以及与上传器交互的主线程的一部分:
......
// Main form
fUniqueID: Integer; // This is a unique number given to each file, because there might be several with the same names(after one is uploaded and deleted)
fUploader: TUploader; // The uploader thread
fFiles: TList<TFileInfo>;
fCurrentFileName: String; // Used to display the progress
function IndexOfFile(aID: Integer): Integer; //Return the index of the record inside the fFiles given the file ID
public
procedure OnCompleteFile(aFileInfo: TFileInfo);
procedure OnUploadError(aError: String);
end;
// This is called by the uploader with Synchronize
procedure TfrmClientMain.OnUploadError(aError: String);
begin
// show and log the error
end;
// This is called by the uploader with Synchronize
procedure TfrmClientMain.OnCompleteFile(aFileInfo: TFileInfo);
var
I: Integer;
begin
I := IndexOfFile(aFileInfo.ID);
if (I >= 0) and (I < fFiles.Count) then
begin
aFileInfo.Status := fsUploaded;
aFileInfo.UploadedSize := aFileInfo.Size;
FFiles.Items[I] := aFileInfo;
Inc(FFilesUploaded);
TFile.Delete(aFileInfo.Path);
colProgressImg.UpdateCell(I);
end;
end;
procedure TfrmClientMain.ProcessFolder;
var
NewFiles: TStringDynArray;
I, J: Integer;
FileInfo: TFileInfo;
begin
// Remove completed files from the list if it contains more than XX files
while FFiles.Count > 1000 do
if FFiles[0].Status = fsUploaded then
begin
Dec(FFilesUploaded);
FFiles.Delete(0);
end else
Break;
NewFiles := TDirectory.GetFiles(WatchFolder, '*.*',TSearchOption.soAllDirectories);
for I := 0 to Length(NewFiles) - 1 do
begin
FileInfo.ID := FUniqueID;
Inc(FUniqueID);
FileInfo.Path := NewFiles[I];
FileInfo.Size := GetFileSizeByName(NewFiles[I]);
FileInfo.UploadedSize := 0;
FileInfo.Status := fsToBeQueued;
FFiles.Add(FileInfo);
if (I mod 100) = 0 then
begin
UpdateStatusLabel;
grFiles.RowCount := FFiles.Count;
Application.ProcessMessages;
if fUploader = nil then
break;
end;
end;
// Send the new files and resend failed to the uploader thread
for I := 0 to FFiles.Count - 1 do
if (FFiles[I].Status = fsToBeQueued) then
begin
if fUploader = nil then
Break;
FileInfo := FFiles[I];
FileInfo.Status := fsQueued;
FFiles[I] := FileInfo;
SaveDebug(1, 'Add: ' + ExtractFileName(FFiles[I].Path));
FUploader.AddFile(FFiles[I]);
end;
end;
procedure TfrmClientMain.tmrGUITimer(Sender: TObject);
var
FileInfo: TFileInfo;
I: Integer;
begin
if (fUploader = nil) or not fUploader.Uploading then
Exit;
FileInfo := fUploader.GetProgress;
I := IndexOfFile(FileInfo.ID);
if (I >= 0) and (I < fFiles.Count) then
begin
fFiles.Items[I] := FileInfo;
fCurrentFileName := ExtractFileName(FileInfo.Path);
colProgressImg.UpdateCell(I);
end;
end;
function TfrmClientMain.IndexOfFile(aID: Integer): Integer;
var
I: Integer;
begin
Result := -1;
for I := 0 to FFiles.Count - 1 do
if FFiles[I].ID = aID then
Exit(I);
end;
这可能不是问题，但 TFileInfo 是一条记录。
这意味着当作为（非 const/var）参数传递时，它会被复制。这可能会导致记录中的字符串等问题，在复制记录时，引用计数不会更新。
要尝试的一件事是使其成为一个类并传递一个实例作为参数（即指向堆上数据的指针）。
其他需要注意的是线程 32 位系统上的共享 Int64（例如您的大小值）。
更新/读取这些内容不是以原子方式完成的，并且您没有任何特定的保护，因此读取值可能会由于线程而导致高位和低位 32 位不匹配。（例如读取高32位，写入高32位，写入低32位，读取低32位，在不同线程中读取和写入）。这可能不会导致您所看到的问题，除非您正在处理 > 4GB 的文件传输，否则不太可能导致任何问题。