Синхронизация/отправка данных между потоками

Приложение написано в Delphi XE.

У меня есть два класса: TBoss и TWorker, которые основаны на TThread. TBoss - это поток одного экземпляра, который запускается, а затем создает около 20 потоков TWorker.

Когда босс создает экземпляр TWorker, он назначает ему метод вызова синхронизации, когда рабочий закончил с тем, что он делает, и вызывает этот метод, который позволяет Боссу получить доступ к записи на Рабочем месте.

Однако я чувствую, что это проблема, вызывающая синхронизация, похоже, блокирует все приложение - блокирует основной (ui) поток. На самом деле нужно просто синхронизировать этого работника с потоком босса....

Раньше я использовал сообщения/упакованные записи для отправки контента между потоками, которые хорошо работали. Однако делать это таким образом намного чище и красивее.... просто очень блокирует.

Есть ли способ вызвать Syncronize в рабочем состоянии, чтобы ждать только поток Boss?

Мой код:

    type 
      TWorker = class(TThread) 
      private 
        fResult : TResultRecord;
        procedure SetOnSendResult(const Value: TNotifyEvent);
        ....
        ....
      public
        property OnSendResult: TNotifyEvent write SetOnSendResult; 
        property Result : TResultRecord read fResult;
        ....
     end;

    ...
    ...
    procedure TWorker.SendBossResults; 
    begin 
      if (Terminated = False) then 
      begin 
        Synchronize(SendResult); 
      end; 
    end; 

    procedure TWorker.SendResult; 
    begin 
      if (Terminated = false) and Assigned(FOnSendResult) then 
      begin 
        FOnSendResult(Self); 
      end; 
    end;

Тогда в моей теме Boss я сделаю что-то вроде этого

    var 
      Worker  : TWorker; 
    begin 
      Worker              := TWorker.Create; 
      Worker.OnTerminate  := OnWorkerThreadTerminate; 
      Worker.OnSendResult := ProcessWorkerResults;

Итак, у моего босса есть метод ProcessWorkerResults - это то, что запускается на Synchronize (SendResult); рабочего.

    procedure TBoss.ProcessWorkerResults(Sender: TObject); 
    begin 
      if terminated = false then 
      begin 
        If TWorker(Sender).Result.HasRecord then
        begin
          fResults.Add(TWorker(Sender).Result.Items);
        end;
      end; 
    end;

Ответы

Ответ 1

Synchronize специально разработан для выполнения кода в потоке main; почему он, кажется, блокирует все.

Вы можете использовать несколько способов связи из рабочих потоков с потоком босса:

  • Добавить обратный вызов для каждого рабочего потока, и назначить его из потока босса когда он создан. Он может вернуться назад как параметры, наряду с идентификатор потока или другой идентификатор.

  • Опубликовать сообщение из рабочего потока к потоку босса, используя PostThreadMessage. Недостатком является то, что босс поток должен иметь ручку окна (см. Classes.AllocateHWnd в Справка Delphi и комментарий Дэвида Хеффернана ниже).

  • Используйте хорошее качество сторонних потоковая библиотека. Видеть OmniThreadLibrary - бесплатно, ОС и очень хорошо написаны.

Мой выбор был бы третьим. Примоз сделал для вас всю тяжелую работу.:)

После вашего комментария, здесь что-то вроде первого предложения. Обратите внимание, что это непроверено, так как написав код для потока TBoss и TWorker + тестовое приложение немного длиннее, когда я прав в эту минуту... Этого должно быть достаточно, чтобы дать вам я надеюсь.

type 
  TWorker = class(TThread) 
  private 
    fResult : TResultRecord;
    fListIndex: Integer;
    procedure SetOnSendResult(const Value: TNotifyEvent);
    ....
    ....
  public
    property OnSendResult: TNotifyEvent write SetOnSendResult; 
    property Result : TResultRecord read fResult;
    property ListIndex: Integer read FListIndex write FListIndex;
    ....
  end;

type 
  TBoss=class(TThread)
  private
    FWorkerList: TThreadList; // Create in TBoss.Create, free in TBoss.Free
    ...
  end;

procedure TWorker.SendBossResults; 
begin 
  if not Terminated then
    SendResult; 
end;

procedure TBoss.ProcessWorkerResults(Sender: TObject); 
var
  i: Integer;
begin 
  if not terminated then 
  begin 
    If TWorker(Sender).Result.HasRecord then
    begin
      FWorkerList.LockList;
      try
        i := TWorker(Sender).ListIndex;
        // Update the appropriate record in the WorkerList
        TResultRecord(FWorkerList[i]).Whatever...
      finally
        FWorkerList.UnlockList;
      end;
    end;
  end; 
end;

Ответ 2

Вы можете использовать поточную безопасную очередь. В DelphiXE есть TThreadedQueue. Если у вас нет DXE, попробуйте OmniThreadLibray - эта библиотека очень хороша для всех проблем с потоками.

Ответ 3

Как я уже упоминал о новых опциях в Delphi 2009 и выше, здесь приведена ссылка на пример для связи между производителями и потребителями между потоками на основе новых блокировок objct в моем блоге:

Синхронизация потоков с защищенными блоками в Delphi

В примечании относительно устаревших методов TThread.Suspend и TThread.Resume, DocWiki для Embarcadero для Delphi рекомендует, чтобы "поток методы синхронизации должны быть на основе SyncObjs.TEvent и SyncObjs.TMutex". Однако есть, другой класс синхронизации доступный с Delphi 2009: TMonitor. Он использует блокировку объекта, которая была представленный в этой версии...

Ответ 4

public свойства класса TWorker ДОЛЖНЫ иметь методы get и set, поэтому вы можете использовать Tcriticalsection для получения значений свойств. В противном случае у вас будут проблемы с потоками. Ваш пример выглядит нормально, но в реальном мире тысячи потоков, обращающихся к одному и тому же значению, приведут к ошибке чтения. Используйте критические разделы.. и вам не нужно будет использовать синхронизацию. Таким образом, вы избегаете перехода в очереди сообщений окон и повышения производительности. Кроме того, если вы используете этот код в приложении для Windows-приложений (там, где не разрешены сообщения Windows), этот пример не будет работать. Метод синхронизации не работает, если нет доступа к очереди сообщений Windows.

Ответ 5

Решено!! (ответ от вопроса)
Исправления сделаны для этой проблемы, где два раза.
Сначала удалите вызов syncronization в методе TWorker SendBossResult.

Второе добавьте класс fProcessWorkerResult CritialSection в TBoss. Создайте и освободите это для создания/уничтожения TBoss. В методе ProcessWorkerResults вызывается метод fProcessWorkerResult.Enter и fProcessWorkerResult.leave вокруг кода, который должен быть безопасным из нескольких потоков рабочих результатов.

Выше было заключение после кода Кенса и последующий комментарий. Большое спасибо, сэр, шлемы для вас!