您可以使用OmniThreadLibrary 并行处理数据库中的记录，而不会带来太多麻烦。
我使用 Pipeline 抽象编写了一个示例。 3个阶段的管道常量：
- 第一阶段从数据库中读取数据，创建一个容器对象的实例来代表管道下一阶段的数据。
-
第二阶段处理传入的数据。
- 调用
DoSomethingWith 过程，这只会浪费大约 100 毫秒。模拟数据的处理
- 释放容器实例的内存。
- 然后将文字值
1 添加到输出队列，以通知最后阶段已处理了另一条记录。
此阶段配置为在 7 个线程中并行运行。
- 最后一个阶段只是统计从前一个阶段完成了多少条记录
该示例是一个控制台应用程序，您只需复制/粘贴即可查看它在您的机器中的实时运行情况。
program Project1;
{$APPTYPE CONSOLE}
{$R *.res}
uses
System.SysUtils,
OtlCommon,
OtlCollections,
OtlParallel,
System.Diagnostics,
DB, DBClient;
type
//auxiliar container, used to copy the database data
//to avoid synchronization. remember TDataSet "current record"
//may cause conflicts if changed from different threads.
TContainer = class
private
FName: string;
FID: Int64;
public
property ID: Int64 read FID write FID;
property Name: string read FName write FName;
end;
//does nothing, but wastes around 100ms. "processing" each record
procedure DoSomethingWith(const AValue: TContainer);
begin
Sleep(100);
end;
//creates a DataSet on the fly with a random number of records
function CreateDataSet: TClientDataSet;
var
I: Integer;
begin
Result := TClientDataSet.Create(nil);
with Result.FieldDefs.AddFieldDef do
begin
Name := 'ID';
DataType := ftLargeint;
end;
with Result.FieldDefs.AddFieldDef do
begin
Name := 'NAME';
DataType := ftString;
end;
Result.CreateDataSet;
for I := 1 to Random(1000) do
Result.InsertRecord([I, 'Test']);
end;
var
RecordsProcessed: Integer;
SW: TStopwatch;
Data: TDataSet;
begin
IsMultiThread := True;
Randomize;
Writeln('wait while processing...');
SW := TStopwatch.Create;
SW.Start;
try
Data := CreateDataSet;
try
RecordsProcessed := Parallel.Pipeline
.Stage(
procedure (const Input, Output: IOmniBlockingCollection)
var
RecData: TContainer;
begin
Data.First;
while not Data.Eof do
begin
RecData := TContainer.Create;
RecData.ID := Data.Fields[0].AsLargeInt;
RecData.Name := Data.Fields[1].AsString;
Output.Add(RecData);
Data.Next;
end;
end)
.Stage(
procedure (const Input: TOmniValue; var Output: TOmniValue)
begin
//process the real thing here
DoSomethingWith(Input);
Input.AsObject.Free;
Output := 1; //another record
end)
.NumTasks(7) //this stage is processed by 7 parallel tasks
.Stage(
procedure (const Input, Output: IOmniBlockingCollection)
var
Recs: Integer;
Value: TOmniValue;
begin
Recs := 0;
for Value in Input do
Inc(Recs, Value);
Output.Add(Recs);
end)
.Run.Output.Next;
SW.Stop;
Writeln(RecordsProcessed, ' records processed in ', SW.ElapsedMilliseconds, 'ms.');
Writeln('Avg. ', (SW.ElapsedMilliseconds/RecordsProcessed):0:3, 'ms./record');
finally
Data.Free;
end;
except
on E: Exception do
Writeln(E.ClassName, ': ', E.Message);
end;
readln;
end.
恕我直言，这样做的主要优点是：
- 您有一个灵活的机制在多个工作人员之间分配工作。如果某些记录需要更多时间来处理，图书馆会处理这种情况，您可以合理地期望在更短的时间内完成全部工作。
- 从数据库中读取第一条记录后，您的第一个处理线程就会立即启动。
- 如果您必须等待基表中的更多传入记录，您可以轻松调整它。在阶段过程中的代码结束之前，阶段的输出队列不会被标记为完成。如果在某个时候没有更多工作要做，所有即将到来的阶段都会阻止等待更多数据处理。
- 只需更改参数值即可更改工作线程数！