附加到先前由另一个线程创建的数据集

Question

这个问题与此类似,但背景不同.

编译器:Delphi 2010,即将推出Delphi XE5.

我已经构建了一个很好的应用程序,通过ZEOS组件管理远程MySQL服务器上的数据.由于连接可能失败并且SQL很慢,我使用了整齐的OmniThreadLibrary来创建一个SQL服务器看门狗并卸载了很多"只读"表加载到线程.截至目前,我在主窗体显示之前手动创建了三个数据模块,每个模块都有独立的TZConnection和一些链接到同一数据模块TZConnection的TZReadOnlyQuery组件.每个线程从其自身内部实例化其相关数据模块,然后执行查询.

看门狗工作得很好,但我对第二部分有疑问,那就是"只读"表格线程.查询已经工作但我还没有在主应用程序业务代码中使用他们的结果,我必须在其他表上插入和更新数据.

在我的计划中,我在主应用程序甚至连接它们之前读取并加载了所有这些"只读"数据集(整个线程间状态机已经完成).理论上应该没有并发问题,因为"只读"表线程已完成其任务并且现在处于空闲状态.但我不知道如果此时我将控件或其他数据集/数据源/从主窗体连接到空闲线程数据模块会发生什么.

我会搞砸,因为主要形式TZSession与线程数据模块不一样吗？只有在交付申请后才能发现罕见且令人讨厌的访问违规行为(当然!).基本上我应该使用什么样的信心或预防措施来访问在另一个线程中创建的查询组件,假设只有主应用程序执行它并且仅用于读取数据？它甚至可能/健康吗？或者我错过了一些"最佳实践"的方式吗？

提前致谢.

Answer 1

我将发布我是如何做到的。由于时间不够，它会变得“简洁”（它仍然是一堵巨大的文字墙！），如果您发现某些内容太晦涩，请随时询问。我不会假装自己没有编写过最好、最快或格式最好的代码。只需使用它作为您的起点即可。

问题领域的简短回顾：拥有一个在启动时使用线程“预打开”表的软件。这使得软件保持响应，甚至执行其他非数据库相关的启动任务。我已经测试了这个程序 1 个月，所使用的数据库组件足以证明它不仅仅是一个“玩具演示”，在您添加第三个数据集时随时可能被破坏。

原料：

如上所述：Delphi 2010+（我现在在 RAD Studio XE5 Ultimate 中运行它），但可能可以与早期版本一起使用。我只是没有测试它们。

• ZEOS 库，7 及以上版本应该可以工作，包括最新的 7.2 alpha。
• 全线程库
• 就我而言，我还使用了 JVCL，但这只是因为我需要一些基本组件不提供的特定数据源事件。

IDE，非代码部分

• 创建将托管数据库组件的 2 个数据模块：一个用于“预加载”、线程部分，另一个用于“运行时”、读写部分，基本上将由程序用户用来执行其任务。

• 创建 1 个单元来存储线程工作代码

• 创建 1 个或多个表格来构成您的申请。

这两个数据模块如下所示：

预加载模块，由工作线程管理。

主数据库模块的一部分。它包括预加载模块无法预加载的几个附加数据集。这些通常是“动态”数据集，其查询直接受到与用户交互的影响。主数据库模块的组件是作为预加载模块的复制和粘贴开始的，因此准备这两个模块不需要花费两倍的时间。

预加载和主数据库模块都附带 ConnectToDatabase 和 DisconnectFromDatabase 过程，这些过程执行系统启动和运行所需的所有步骤。

很重要！

• 预加载模块在单独的线程中执行“真实”查询并填充其相关的 TClientDataSet。它的组件没有附加事件。我只将它们用作盲目的“静态”数据容器。
• 主数据库模块将仅“附加”到预加载模块组件。

在示例中：而预加载模块 cdsProducts ClientDataSet 执行“真正的数据库查询”

cdsProduct => dspProduct => qryProduct

链中，主数据库模块 cdsProduct 仅获取预加载模块的 cdsProduct 数据，根本不执行任何查询（否则执行查询两次有什么意义？）。

您会看到，主数据库模块 cdsProduct 还附带链接的 TDataSetProvider 和查询组件，这与直觉是多么反直觉。为什么？因为我用它们来写回修改后的数据。

也就是说，我们的计划分为三个阶段：

1. 启动时，预加载数据模块执行查询（在线程中），仅此而已。没有管理任何事件，所有事件都是只读的。

2. 运行开始阶段，主数据库模块（在VCL线程中）将1中收集的数据复制到其ClientDataSets中。

3. 运行阶段，用户与主数据库模块的 ClientDataSet 交互。当他们需要保存数据时（并且只有在那时），主数据库模块的 DataSetProvider 和查询才会被调用。他们只会写。

对于其中一些 ClientDataSet，我本可以跳过整个 ClientDataSet => Provider => 查询链，但大多数都需要一些大量的数据处理，必须手动更新许多连接表等，所以我只使用了完整的堆栈。

代码部分

让我们来了解一些更具体的细节。我无法发布全部内容，因为它是一个商业应用程序，所以我只会粘贴一些重要的片段。

线程化、预加载数据模块

procedure TModDBPreload.ConnectToDatabase;
begin
    dbcEShop.Connect;
    SendStatusMessage('Loading languages archive');
    qryLanguage.Open;
    qryLanguage.First;
    SearchOptions := [loCaseInsensitive];
    ModApplicationCommon.ApplicationLocaleInfo.Lock;

    ...

    try
        ...

         // All the queries parameters needing a language id need to be assigned to the locked LocaleInfo object
        qryGeoZone.Params.ParamByName('language_id').AsInteger := ModApplicationCommon.ApplicationLocaleInfo.LocaleIDForQueries;
        cdsGeoZones.Params.ParamByName('language_id').AsInteger := ModApplicationCommon.ApplicationLocaleInfo.LocaleIDForQueries;

        ...

    finally
        ModApplicationCommon.ApplicationLocaleInfo.Unlock;
    end;

  SendStatusMessage('Loading countries archive');
  cdsGeoZones.Open;
      cdsGeoZones.First;
  SendStatusMessage('Loading currencies archive');
  qryCurrency.Open;
  qryCurrency.First;
  Sleep(100);
  SendStatusMessage('Loading products archive');
  cdsProduct.Open;
  cdsProduct.First;
  ...
end;

上面的代码片段可以使用很多解释。尤其：

SendStatusMessage('Loading languages archive');

是一个发送最终用户友好的更新字符串以显示在状态行上的线程。当然状态行是由主VCL线程管理的。怎么做？我稍后会展示它。

qryLanguage.Open;
qryLanguage.First;
...
cdsGeoZones.Open;
cdsGeoZones.First;

并非所有数据集都需要在整个应用程序持续时间内进行管理。只有那些需要的数据才由 ClientDataSets 管理。发生“第一次”调用是因为我不知道服务器后端是否会改变。某些数据库驱动程序、DLL、（特别是）ODBC 连接器等不会在打开期间执行实际的繁重工作，而是在第一个游标操作时执行。因此，即使当前的驱动程序并不严格需要它，我也会确保它发生。这Sleep(100)是为了让用户和开发人员在打开小表时看到消息。当然，一旦软件最终完成，就可能被删除。

这些Lock, try / finally条款等是为了提醒您，我们处于一个线程中，并且最好在采取一些预防措施的情况下访问某些资源。在这种特定情况下，我们还有其他线程（与本文无关，因此未涵盖），因此我们必须保护一些数据结构。特别是，我“借用”了基本的 Delphi 线程安全列表锁定机制 paradygm，因此方法名称也是相同的。

基于 OmniThreadLibrary 的预加载模块线程工作者

这是最相关/教学代码：

type
  TDBPreloadWorker = class(TOmniWorker)
  protected
    ThreadModDatabase : TModDBPreload;
    FStatusString : string;
  public
    constructor Create;
    function Initialize : boolean; override;
    procedure Cleanup; override;
    procedure SendStatusMessage(anID : Word; aValue : string = ''); overload;
    procedure SendStatusMessage(aValue : string); overload;
    procedure DisconnectFromDatabase;

    procedure OMSendMessage(var msg: TOmniMessage); message MSG_SEND_MESSAGE;
    procedure OMDisconnectFromDatabase(var msg: TOmniMessage); message MSG_DISCONNECT_FROM_DATABASE;
    procedure OMUpdateStateMachine(var msg: TOmniMessage); message MSG_UPDATE_STATE_MACHINE;
  end;
...

constructor TDBPreloadWorker.Create;
begin
  Inherited;
  FStatusString := 'Connecting to server...';
  ThreadModDatabase := Nil;
end;

function TDBPreloadWorker.Initialize : boolean;
begin
  ThreadModDatabase := TModDBPreload.Create(Nil);
  ModDBPreload := ThreadModDatabase;
  ThreadModDatabase.DBPreloadWorker := Self;
  DisconnectFromDatabase; // In case of leftover Active := true from designing the software
  Result := true;
end;

procedure TDBPreloadWorker.Cleanup;
begin
  DisconnectFromDatabase;
  ThreadModDatabase.Free;
  ThreadModDatabase := Nil;
end;

procedure TDBPreloadWorker.SendStatusMessage(anID : Word; aValue : string);
begin
  FStatusString := aValue; // Stored in case the main application polls a status update
  Task.Comm.Send(anID, aValue);
end;

procedure TDBPreloadWorker.SendStatusMessage(aValue : string);
begin
  SendStatusMessage(MSG_GENERAL_RESPONSE, aValue);
end;

procedure TDBPreloadWorker.DisconnectFromDatabase;
begin
  if Assigned(ThreadModDatabase) then
    ThreadModDatabase.DisconnectFromDatabase;
end;

procedure TDBPreloadWorker.OMSendMessage(var msg: TOmniMessage);
begin
  Task.Comm.Send(MSG_GENERAL_RESPONSE, FStatusString);
end;

procedure TDBPreloadWorker.OMDisconnectFromDatabase(var msg: TOmniMessage);
begin
  ...
  DisconnectFromDatabase;
end;

procedure TDBPreloadWorker.OMSendMessage(var msg: TOmniMessage);
begin
  Task.Comm.Send(MSG_GENERAL_RESPONSE, FStatusString);
end;

procedure TDBPreloadWorker.OMUpdateStateMachine(var msg: TOmniMessage);
begin
  Task.Comm.Send(MSG_GENERAL_RESPONSE, FStatusString); // Needed to show the pre-loaded status

  if Assigned(ThreadModDatabase) then
  begin
    try
      ThreadModDatabase.ConnectToDatabase;
      SendStatusMessage('Reading database tables...');

      if not ThreadModDatabase.QueryExecute then
      begin
        raise Exception.Create('Consistency check: the database does not return the expected values');
      end;

      SendStatusMessage(MSG_SUCCESS, 'Tables have been succesfully read');
      SendStatusMessage(MSG_TASK_COMPLETED);

    except
      On E : Exception do
      begin
        DisconnectFromDatabase;
        SendStatusMessage(MSG_TASK_FAILURE, E.Message);
      end;
    end;
  end;
end;

有些代码值得进一步解释：

function TDBPreloadWorker.Initialize : boolean;

创建预加载数据模块。也就是说，所有内容都独立包含在线程的上下文中，并且不会与其他内容发生冲突。

procedure TDBPreloadWorker.SendStatusMessage(anID : Word; aValue : string);

这是如何通过 OmniThreadLibrary 向主 VCL 线程发送消息（顺便说一句，它不限于字符串）。

procedure TDBPreloadWorker.OMUpdateStateMachine(var msg: TOmniMessage);

这是主要的预加载数据模块初始化管理代码。它与 VCL 主线程执行握手，并且基本上充当我在程序中实现的状态机之一。

对于那些想知道所有这些常量从何而来的人：它们是在所有线程相关类包含的单独文件中声明的。它们很简单，可以自由选择整数：

const
  MSG_GENERAL_RESPONSE         = 0;
  MSG_SEND_MESSAGE             = 1;
  MSG_SHUTDOWN                 = 2;
  MSG_SUCCESS                  = $20;
  MSG_ABORT                    = $30;
  MSG_RETRY                    = $31;
  MSG_TASK_COMPLETED           = $40;
  MSG_FAILURE                  = $8020;
  MSG_ABORTED                  = $8030;
  MSG_TASK_FAILURE             = $8040;
  MSG_UPDATE_STATE_MACHINE     = 9;
  MSG_TIMER_1                  = 10;
  MSG_DISCONNECT_FROM_DATABASE = 99;

主窗体端预加载管理代码

各种线程在程序启动时生成。TOmniEventMonitor 的OnTaskMessage事件指向：

procedure TFrmMain.monDBPreloadTaskMessage(const task: IOmniTaskControl;
  const msg: TOmniMessage);
var
  MessageString : string;
  ComponentsNewState : boolean;

begin
  MessageString := msg.MsgData.AsString;

  if Length(MessageString) > 0 then
    UpdateStatusBar(MessageString);

  if task = FDBPreloadWorkerControl then
  begin
    if (msg.MsgID = MSG_TASK_COMPLETED) or (msg.MsgID = MSG_TASK_FAILURE) then
    begin
      ComponentsNewState := (msg.MsgID = MSG_TASK_COMPLETED);

      // Unlike for the watchdog, the preload thread is not terminated
      // The data is needed by the program till its end
      // DBPreloadTerminate;

      // Lets the main database queries be started
      DBPreloadSuccess := (msg.MsgID = MSG_TASK_COMPLETED);
      MainViewEnabled := ComponentsNewState;

      if msg.MsgID = MSG_TASK_FAILURE then
      begin
        if MessageDlg('Unable to load the data tables from the database server', mtError, [mbRetry, mbAbort], 0) = mrAbort then
          Close
        else
          // Reinitialize the preload thread.
          ...
      end;
    end;
  end;
end;

这是一个完全简单的过程，最终会被调用以更新主窗体的状态栏：

procedure TFrmMain.UpdateStatusBar(Value : string);
begin
  pnlStatusBar.SimpleText := Value;
  pnlStatusBar.Update;
  Application.ProcessMessages;
end;

主数据库模块管理代码

最后但并非最不重要的一点是，以下是如何实际“附加”到预加载数据模块 ClientDataSets。从主窗体调用此代码，您的应用程序的基础就基本完成了！

procedure TModDatabase.ConnectToDatabase;

  procedure ConnectDataSet(CDS : TClientDataSet; PreloadDataSet : TClientDataSet; RuntimeDataSet : TZAbstractRODataset; SetLanguage : boolean = false);
  begin
   // Only required by datasets needing a locale_id parameter
    if (SetLanguage) then
    begin
      CDS.Params.ParamByName('language_id').AsInteger := ModApplicationCommon.ApplicationLocaleInfo.LocaleIDForQueries;
      RuntimeDataSet.ParamByName('language_id').AsInteger := ModApplicationCommon.ApplicationLocaleInfo.LocaleIDForQueries;
    end;

    CDS.Data := PreloadDataSet.Data;
    CDS.Active := true;
  end;

begin
  DisconnectFromDatabase;
  dbcEShop.Connect;

  UpdateStatusBar('Setting up products archive');
  ConnectDataSet(cdsProduct, ModDBPreload.cdsProduct, qryProduct, true);
  UpdateStatusBar('Setting up products options archive');
  ConnectDataSet(cdsProductOption, ModDBPreload.cdsProductOption, qryProductOption);
  UpdateStatusBar('Setting up options archive');
  ConnectDataSet(cdsOption, ModDBPreload.cdsOption, qryOption);
  UpdateStatusBar('Setting up options descriptions archive');
  ConnectDataSet(cdsOptionDescription, ModDBPreload.cdsOptionDescription, qryOptionDescription, true);
  ...

我希望发布了足够的信息来提供有关整个过程的想法。请随时提出任何问题，对词典感到抱歉，英语是我的第四语言。