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

Question

这是一个类似于 this one 的问题但背景不同。

编译器:Delphi 2010，即将推出Delphi XE5。

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

看门狗工作得很好，但我对第二部分有一些疑问，即“只读”表线程。查询已经工作，但我还没有在主应用程序业务代码中使用它们的结果，我必须在其他表上插入和更新数据。

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

我会不会因为主窗体 TZSession 与线程数据模块不一样而搞砸？我是否会在提交应用程序后才会发现罕见且令人讨厌的访问冲突(当然!)。基本上，我应该以什么样的信心或预防措施访问在另一个线程中创建的查询组件，假设只有主应用程序这样做并且仅用于读取数据？它甚至可能/健康吗？还是我错过了一些“最佳实践”的做法？

提前致谢。

Answer 1

我将发布我是如何做到的。由于时间不够，这将是“简洁”的(它仍然是一个巨大的文字墙!)，如果您发现某些东西太晦涩，请随时提问。我不会假装自己写的既不是最好的，也不是最快的，也不是格式最好的代码。只需将其用作起点即可。

问题域的简短回顾:有一个软件可以在启动时使用线程“预打开”表。这让软件保持响应，甚至执行其他与数据库无关的启动任务。
我已经对这个程序进行了 1 个月的测试，所使用的数据库组件足以证明它不仅仅是一个“玩具演示”，在您添加第三个数据集时会崩溃。

原料:

如上所述:Delphi 2010+(我现在在 RAD Studio XE5 Ultimate 中运行它)但可能适用于早期版本。我只是没有测试它们。

ZEOS 库，7 及更高版本应该可以工作，包括最新的 7.2 alpha。

OmniThreadLibrary

就我而言，我也使用了 JVCL，但这只是因为我需要一些基本组件不提供的特定数据源事件。

IDE，非代码部分

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

创建 1 个单元来存储线程 worker 代码

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

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

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



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



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

很重要!

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

主数据库模块只会“附加”到预加载模块组件。

在示例中:而预加载模块 cdsProducts ClientDataSet 使用

cdsProduct => dspProduct => qryProduct

链，主数据库模块 cdsProduct 只需要预加载模块的 cdsProduct 数据 根本不执行任何查询 (否则有什么意义，执行两次查询？)。

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

也就是说，我们有三个程序阶段:

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

运行开始阶段，主数据库模块(在 VCL 线程中)将 1 中收集的数据复制到其 ClientDataSet 中。

运行阶段，用户与主数据库模块的 ClientDataSet 进行交互。当他们需要保存数据时(并且仅在那时)，主数据库模块的 DataSetProviders 和查询就会参与。他们只会写。

对于其中一些 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 的预加载模块线程 worker

这是最相关/教学代码:

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);
  ...

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