multithreading - 多线程访问数据

Question

我有一个我写的应用程序，我有三个线程。线程 1、2 和 3。

还有两个数字 1 和 2 的列表。

1 号线程将数据插入到 1 号列表中。

2 号线程从列表 1 中获取数据。

2 号线程将数据插入到 2 号列表中。

3 号线程从列表 2 中获取数据。

如果线程同时运行会占用大量CPU。可以在列表中输入数据，不需要第二和第三运行线程。

如何通知线程列表中的新数据被插入并且他们需要开始对新数据进行处理呢？

联系您的技术和有效方式？

非常感谢，非常感谢。

score 2 · Accepted Answer

这是一个示例，它从您的三个线程中添加 list1 和 list2 中的值。

每次将值放入列表时，Event都会触发 an 并且处理此事件的线程会拉出列表中的最后一个值并清除事件标志。

在清除事件标志之前，不能将新值放入列表中。

中间线程对新值进行中间存储，而不是停止第一个线程。

所有事件都是可等待的，因此可以让 cpu 放心。

列表是线程安全的。

program Project62;

{$APPTYPE CONSOLE}

uses
  System.SysUtils,
  System.Classes,
  System.SyncObjs,
  System.Generics.Collections;

Type
  TMyThread1 = Class(TThread)
    private
      fMySyncAddList : TSimpleEvent;
      fMyList : TThreadList<Integer>;
      fAddVal : Integer;
    public
      Constructor Create(ASyncAddList: TSimpleEvent; AList: TThreadList<Integer>);
      procedure Execute; override;
  End;
  TMyThread2 = Class(TThread)
    private
      fMySyncAddList1,fMySyncAddList2 : TSimpleEvent;
      fMyList1,fMyList2 : TThreadList<Integer>;
      fAddVal : Integer;
    public
      Constructor Create(ASyncAddList1,ASyncAddList2: TSimpleEvent; AList1,AList2 : TThreadList<Integer>);
      procedure Execute; override;
  End;
  TMyThread3 = Class(TThread)
    private
      fMySyncAddList2 : TSimpleEvent;
      fMyList2 : TThreadList<Integer>;
      fAddVal : Integer;
    public
      Constructor Create(ASyncAddList2: TSimpleEvent; AList2 : TThreadList<Integer>);
      procedure Execute; override;
  End;


{ TMyThread1 }

constructor TMyThread1.Create( ASyncAddList : TSimpleEvent; AList: TThreadList<Integer>);
begin
  Inherited Create(false);
  fMySyncAddList := AsyncAddList;
  fMyList := AList;

end;

procedure TMyThread1.Execute;
var
  stateAcknowledged : boolean;
begin
  stateAcknowledged := true;
  while (not Terminated) do
  begin
    if stateAcknowledged then
    begin // Do some work and adda a value to list1
      fAddVal := Random(100);
      fMyList.Add(fAddVal);
      fMySyncAddList.SetEvent; // Signal a new addition
      stateAcknowledged := false;
      //ShowVal;
      Sleep(1000);
    end
    else begin
      stateAcknowledged := (fMySyncAddList.WaitFor(100) <> wrSignaled);
    end;
  end;
end;

{ TMyThread2 }

constructor TMyThread2.Create(ASyncAddList1, ASyncAddList2: TSimpleEvent;
  AList1, AList2: TThreadList<Integer>);
begin
  Inherited Create(false);
  fMySyncAddList1 := AsyncAddList1;
  fMySyncAddList2 := AsyncAddList2;
  fMyList1 := AList1;
  fMyList2 := AList2;    
end;

procedure TMyThread2.Execute;
var
  wr : TWaitResult;
  list : TList<Integer>;
  pulled : Boolean;
begin
  pulled := false;
  while (not Terminated) do
  begin
    if pulled then // Add a value to list2
    begin
      wr := fMySyncAddList2.WaitFor(0);
      if (wr <> wrSignaled) then
      begin
        fMyList2.Add(fAddVal);
        fMySyncAddList2.SetEvent; // Signal a new addition
        pulled := false;
      end
      else Sleep(100);
    end
    else begin // Wait for a new value in list1
      wr := fMySyncAddList1.WaitFor(INFINITE);
      if Terminated then
        Exit;
      if (wr = wrSignaled) then
      begin
        // Pull out the value
        list := fMyList1.LockList;
        try
          fAddVal := list.Last;
        finally
          fMyList1.UnlockList;
        end;
        // All clear
        pulled := true;
        fMySyncAddList1.ResetEvent;
        //ShowVal;
      end;
    end;
  end;
end;

{ TMyThread3 }

constructor TMyThread3.Create(ASyncAddList2: TSimpleEvent;
  AList2: TThreadList<Integer>);
begin
  Inherited Create(false);
  fMySyncAddList2 := AsyncAddList2;
  fMyList2 := AList2;
end;

procedure TMyThread3.Execute;
var
  wr : TWaitResult;
  list : TList<Integer>;
begin
  while not Terminated do
  begin
    wr := fMySyncAddList2.WaitFor(INFINITE);
    if Terminated then
      Exit;
    if (wr = wrSignaled)  then // Wait for signal
    begin
      // Pull out the value
      list := fMyList2.LockList;
      try
        fAddVal := list.Last;
        //ShowVal;
      finally
        fMyList2.UnlockList;
      end;
      // Clear event
      fMySyncAddList2.ResetEvent;
    end;
  end;
end;

var
  list1,list2 : TThreadList<Integer>;
  syncList1,syncList2 : TSimpleEvent;
  thread1 : TMyThread1;
  thread2 : TMyThread2;
  thread3 : TMyThread3;
begin
  list1 := TThreadList<Integer>.Create;
  list2 := TThreadList<Integer>.Create;
  syncList1 := TSimpleEvent.Create(Nil,True,False,'',false);
  syncList2 := TSimpleEvent.Create(Nil,True,False,'',false);
  thread3 := TMyThread3.Create(syncList2,list2);
  thread2 := TMyThread2.Create(syncList1,syncList2,list1,list2);
  thread1 := TMyThread1.Create(syncList1,list1);
  Try
    WriteLn('Press [Enter] key to stop.');
    ReadLn;

  Finally
    thread3.Terminate;
    syncList2.SetEvent; // Wake up call
    thread3.Free;
    thread2.Terminate;
    syncList1.SetEvent; // Wake up call
    thread2.Free;
    thread1.Free;
    syncList1.Free;
    syncList2.Free;
    list1.Free;
    list2.Free;
  End;
end.

添加了两个TThreadedQueue在线程之间传输信息的示例。线程改为保留整数的内部列表。正如@DavidHeffernan 指出的那样，代码要简单得多。

program Project63;

{$APPTYPE CONSOLE}

uses
  System.SysUtils,
  System.Classes,
  System.SyncObjs,
  System.Generics.Collections;

Type
  TMyThread1 = Class(TThread)
    private
      fMyList : TList<Integer>;
      fMyQueue : TThreadedQueue<Integer>;
      fAddVal : Integer;
    public
      Constructor Create(AQueue : TThreadedQueue<Integer>);
      procedure Execute; override;
  End;
  TMyThread2 = Class(TThread)
    private
      fMyList1,fMyList2 : TList<Integer>;
      fMyQueue1,fMyQueue2 : TThreadedQueue<Integer>;
      fAddVal : Integer;
    public
      Constructor Create(AQueue1,AQueue2: TThreadedQueue<Integer>);
      procedure Execute; override;
  End;
  TMyThread3 = Class(TThread)
    private
      fMyList : TList<Integer>;
      fMyQueue : TThreadedQueue<Integer>;
      fAddVal : Integer;
    public
      Constructor Create(AQueue : TThreadedQueue<Integer>);
      procedure Execute; override;
  End;

constructor TMyThread1.Create( AQueue : TThreadedQueue<Integer>);
begin
  Inherited Create(false);
  fMyQueue:= AQueue;
  fMyList := TList<Integer>.Create;
end;

procedure TMyThread1.Execute;
begin
  while (not Terminated) do
  begin
    Sleep(1000); // Simulate some work
    fAddVal := Random(100);
    fMyList.Add(fAddVal);
    fMyQueue.PushItem(fAddVal); // Signal a new addition
  end;
  fMyList.Free;
end;

constructor TMyThread2.Create(AQueue1,AQueue2: TThreadedQueue<Integer>);
begin
  Inherited Create(false);
  fMyQueue1 := AQueue1;
  fMyQueue2 := AQueue2;
  fMyList1 := TList<Integer>.Create;
  fMyList2 := TList<Integer>.Create;
end;

procedure TMyThread2.Execute;
var
  queueSize : Integer;
begin
  while (not Terminated) do
  begin
    if (fMyQueue1.PopItem(queueSize,fAddVal) = wrSignaled) and 
       (not Terminated) then
    begin
      fMyList1.Add(fAddVal);
      // Do some work and send a new value to next thread
      fMyQueue2.PushItem(fAddVal);
      fMyList2.Add(fAddVal);
      
    end;
  end;
  fMyList1.Free;
  fMyList2.Free;
end;

constructor TMyThread3.Create(AQueue : TThreadedQueue<Integer>);
begin
  Inherited Create(false);
  fMyQueue := AQueue;
  fMyList := TList<Integer>.Create;
end;

procedure TMyThread3.Execute;
var
  queueSize : Integer;
begin
  while not Terminated do
  begin
    if (fMyQueue.PopItem(queueSize,fAddVal) = wrSignaled) and 
       (not Terminated) then
    begin
      fMyList.Add(fAddVal);
      // Do some work on list
      
    end;
  end;
  fMyList.Free;
end;

var
  queue1,queue2 : TThreadedQueue<Integer>;
  thread1 : TMyThread1;
  thread2 : TMyThread2;
  thread3 : TMyThread3;
begin
  queue1 := TThreadedQueue<Integer>.Create;
  queue2 := TThreadedQueue<Integer>.Create;
  thread3 := TMyThread3.Create(queue2);
  thread2 := TMyThread2.Create(queue1,queue2);
  thread1 := TMyThread1.Create(queue1);
  Try
    WriteLn('Press [Enter] key to stop.');
    ReadLn;

  Finally
    thread3.Terminate;
    queue2.PushItem(0); // Wake up call
    thread3.Free;
    thread2.Terminate;
    queue1.PushItem(0); // Wake up call
    thread2.Free;
    thread1.Free;
    queue1.Free;
    queue2.Free;
  End;
end.

score 1 · Accepted Answer

首先想到的是信号量。信号量是一种内部带有计数器的同步原语：一个线程尝试递减计数器，如果它为零，则线程阻塞（即没有调度并且不占用 CPU，只是无害地等待）。另一个线程增加计数器，导致阻塞线程继续。

您可能对每个列表都有一个信号量，因此消费线程在读取之前递减信号量，而生产者线程在写入后递增它。

还有一件事需要注意：是否允许一个线程从列表中获取数据，而另一个线程正在修改它？这取决于列表（或一般的集合）的实现方式，因此请尝试在文档中查找有关列表的“线程安全”的内容。也许您需要另一个同步 primite：互斥锁。互斥锁被线程“获取”然后“释放”，当多个线程试图锁定一个互斥锁时，其中一个必须等待。每个列表一个互斥锁，在此互斥锁上进行修改和读取，在这里可能是合适的。

multithreading - 多线程访问数据

2 回答 2

Related

Reference