TCommThread--在delphi线程中实现消息循环

原作者: [db:作者] 来自: [db:来源] 收藏邀请

http://www.techques.com/question/1-4073197/How-do-I-send-and-handle-message-between-TService-parent-thread-and-child-thread?

I took a look at OmniThreadLibrary and it looked like overkill for my purposes.

I wrote a simple library I call TCommThread.

It allows you to pass data back to the main thread without worrying about

any of the complexities of threads or Windows messages.

Here's the code if you'd like to try it.

CommThread Library:

  1 unit Threading.CommThread;
  2 
  3 interface
  4 
  5 uses
  6   Classes, SysUtils, ExtCtrls, SyncObjs, Generics.Collections, DateUtils;
  7 
  8 const
  9   CTID_USER = 1000;
 10   PRM_USER = 1000;
 11 
 12   CTID_STATUS = 1;
 13   CTID_PROGRESS = 2;
 14 
 15 type
 16   TThreadParams = class(TDictionary<String, Variant>);
 17   TThreadObjects = class(TDictionary<String, TObject>);
 18 
 19   TCommThreadParams = class(TObject)
 20   private
 21     FThreadParams: TThreadParams;
 22     FThreadObjects: TThreadObjects;
 23   public
 24     constructor Create;
 25     destructor Destroy; override;
 26 
 27     procedure Clear;
 28 
 29     function GetParam(const ParamName: String): Variant;
 30     function SetParam(const ParamName: String; ParamValue: Variant): TCommThreadParams;
 31     function GetObject(const ObjectName: String): TObject;
 32     function SetObject(const ObjectName: String; Obj: TObject): TCommThreadParams;
 33   end;
 34 
 35   TCommQueueItem = class(TObject)
 36   private
 37     FSender: TObject;
 38     FMessageId: Integer;
 39     FCommThreadParams: TCommThreadParams;
 40   public
 41     destructor Destroy; override;
 42 
 43     property Sender: TObject read FSender write FSender;
 44     property MessageId: Integer read FMessageId write FMessageId;
 45     property CommThreadParams: TCommThreadParams read FCommThreadParams write FCommThreadParams;
 46   end;
 47 
 48   TCommQueue = class(TQueue<TCommQueueItem>);
 49 
 50   ICommDispatchReceiver = interface
 51     ['{A4E2C9D1-E4E8-497D-A9BF-FAFE2D3A7C49}']
 52     procedure QueueMessage(Sender: TObject; MessageId: Integer; CommThreadParams: TCommThreadParams);
 53     procedure CommThreadTerminated(Sender: TObject);
 54     function Cancelled: Boolean;
 55   end;
 56 
 57   TCommThread = class(TThread)
 58   protected
 59     FCommThreadParams: TCommThreadParams;
 60     FCommDispatchReceiver: ICommDispatchReceiver;
 61     FName: String;
 62     FProgressFrequency: Integer;
 63     FNextSendTime: TDateTime;
 64 
 65     procedure SendStatusMessage(const StatusText: String; StatusType: Integer = 0); virtual;
 66     procedure SendProgressMessage(ProgressID: Int64; Progress, ProgressMax: Integer; AlwaysSend: Boolean = TRUE); virtual;
 67   public
 68     constructor Create(CommDispatchReceiver: TObject); reintroduce; virtual;
 69     destructor Destroy; override;
 70 
 71     function SetParam(const ParamName: String; ParamValue: Variant): TCommThread;
 72     function GetParam(const ParamName: String): Variant;
 73     function SetObject(const ObjectName: String; Obj: TObject): TCommThread;
 74     function GetObject(const ObjectName: String): TObject;
 75     procedure SendCommMessage(MessageId: Integer; CommThreadParams: TCommThreadParams); virtual;
 76 
 77     property Name: String read FName;
 78   end;
 79 
 80   TCommThreadClass = Class of TCommThread;
 81 
 82   TCommThreadQueue = class(TObjectList<TCommThread>);
 83 
 84   TCommThreadDispatchState = (
 85     ctsIdle,
 86     ctsActive,
 87     ctsTerminating
 88   );
 89 
 90   TOnReceiveThreadMessage = procedure(Source, Sender: TObject; MessageId: Integer; CommThreadParams: TCommThreadParams) of object;
 91   TOnStateChange = procedure(Sender: TObject; State: TCommThreadDispatchState) of object;
 92   TOnStatus = procedure(Source, Sender: TObject; const ID: String; StatusText: String; StatusType: Integer) of object;
 93   TOnProgress = procedure(Source, Sender: TObject; const ID: String; Progress, ProgressMax: Integer) of object;
 94 
 95   TBaseCommThreadDispatch = class(TComponent, ICommDispatchReceiver)
 96   private
 97     FProcessQueueTimer: TTimer;
 98     FCSReceiveMessage: TCriticalSection;
 99     FCSCommThreads: TCriticalSection;
100     FCommQueue: TCommQueue;
101     FActiveThreads: TList;
102     FCommThreadClass: TCommThreadClass;
103     FCommThreadDispatchState: TCommThreadDispatchState;
104 
105     function CreateThread(const ThreadName: String = ''): TCommThread;
106     function GetActiveThreadCount: Integer;
107     function GetStateText: String;
108   protected
109     FOnReceiveThreadMessage: TOnReceiveThreadMessage;
110     FOnStateChange: TOnStateChange;
111     FOnStatus: TOnStatus;
112     FOnProgress: TOnProgress;
113     FManualMessageQueue: Boolean;
114     FProgressFrequency: Integer;
115 
116     procedure SetManualMessageQueue(const Value: Boolean);
117     procedure SetProcessQueueTimerInterval(const Value: Integer);
118     procedure SetCommThreadDispatchState(const Value: TCommThreadDispatchState);
119     procedure QueueMessage(Sender: TObject; MessageId: Integer; CommThreadParams: TCommThreadParams);
120     procedure OnProcessQueueTimer(Sender: TObject);
121     function GetProcessQueueTimerInterval: Integer;
122 
123     procedure CommThreadTerminated(Sender: TObject); virtual;
124     function Finished: Boolean; virtual;
125 
126     procedure DoOnReceiveThreadMessage(Sender: TObject; MessageId: Integer; CommThreadParams: TCommThreadParams); virtual;
127     procedure DoOnStateChange; virtual;
128 
129     procedure TerminateActiveThreads;
130 
131     property OnReceiveThreadMessage: TOnReceiveThreadMessage read FOnReceiveThreadMessage write FOnReceiveThreadMessage;
132     property OnStateChange: TOnStateChange read FOnStateChange write FOnStateChange;
133     property OnStatus: TOnStatus read FOnStatus write FOnStatus;
134     property OnProgress: TOnProgress read FOnProgress write FOnProgress;
135 
136     property ProgressFrequency: Integer read FProgressFrequency write FProgressFrequency;
137     property ProcessQueueTimerInterval: Integer read GetProcessQueueTimerInterval write SetProcessQueueTimerInterval;
138     property ManualMessageQueue: Boolean read FManualMessageQueue write SetManualMessageQueue;
139     property CommThreadDispatchState: TCommThreadDispatchState read FCommThreadDispatchState write SetCommThreadDispatchState;
140   public
141     constructor Create(AOwner: TComponent); override;
142     destructor Destroy; override;
143 
144     function NewThread(const ThreadName: String = ''): TCommThread; virtual;
145     procedure ProcessMessageQueue; virtual;
146     procedure Stop; virtual;
147     function State: TCommThreadDispatchState;
148     function Cancelled: Boolean;
149 
150     property ActiveThreadCount: Integer read GetActiveThreadCount;
151     property StateText: String read GetStateText;
152 
153     property CommThreadClass: TCommThreadClass read FCommThreadClass write FCommThreadClass;
154   end;
155 
156   TCommThreadDispatch = class(TBaseCommThreadDispatch)
157   published
158     property OnReceiveThreadMessage: TOnReceiveThreadMessage read FOnReceiveThreadMessage write FOnReceiveThreadMessage;
159     property OnStateChange: TOnStateChange read FOnStateChange write FOnStateChange;
160 
161     property ProgressFrequency: Integer read FProgressFrequency write FProgressFrequency;
162     property ProcessQueueTimerInterval: Integer read GetProcessQueueTimerInterval write SetProcessQueueTimerInterval;
163     property ManualMessageQueue: Boolean read FManualMessageQueue write SetManualMessageQueue;
164   end;
165 
166   TBaseStatusCommThreadDispatch = class(TBaseCommThreadDispatch)
167   protected
168     FOnStatus: TOnStatus;
169     FOnProgress: TOnProgress;
170 
171     procedure DoOnReceiveThreadMessage(Sender: TObject; MessageId: Integer; CommThreadParams: TCommThreadParams); override;
172 
173     procedure DoOnStatus(Sender: TObject;const ID: String; const StatusText: String; StatusType: Integer); virtual;
174     procedure DoOnProgress(Sender: TObject; const ID: String; Progress, ProgressMax: Integer); virtual;
175 
176     property OnStatus: TOnStatus read FOnStatus write FOnStatus;
177     property OnProgress: TOnProgress read FOnProgress write FOnProgress;
178   end;
179 
180   TStatusCommThreadDispatch = class(TBaseStatusCommThreadDispatch)
181   published
182     property OnReceiveThreadMessage: TOnReceiveThreadMessage read FOnReceiveThreadMessage write FOnReceiveThreadMessage;
183     property OnStateChange: TOnStateChange read FOnStateChange write FOnStateChange;
184     property OnStatus: TOnStatus read FOnStatus write FOnStatus;
185     property OnProgress: TOnProgress read FOnProgress write FOnProgress;
186 
187     property ProgressFrequency: Integer read FProgressFrequency write FProgressFrequency;
188     property ProcessQueueTimerInterval: Integer read GetProcessQueueTimerInterval write SetProcessQueueTimerInterval;
189     property ManualMessageQueue: Boolean read FManualMessageQueue write SetManualMessageQueue;
190   end;
191 
192 implementation
193 
194 const
195   PRM_STATUS_TEXT = 'Status';
196   PRM_STATUS_TYPE = 'Type';
197   PRM_PROGRESS_ID = 'ProgressID';
198   PRM_PROGRESS = 'Progess';
199   PRM_PROGRESS_MAX = 'ProgressMax';
200 
201 resourcestring
202   StrCommReceiverMustSupportInterface = 'CommDispatchReceiver must support ICommDispatchReceiver interface';
203   StrSenderMustBeATCommThread = 'Sender must be a TCommThread';
204   StrUnableToFindTerminatedThread = 'Unable to find the terminated thread';
205   StrIdle = 'Idle';
206   StrTerminating = 'Terminating';
207   StrActive = 'Active';
208 
209 { TCommThread }
210 
211 constructor TCommThread.Create(CommDispatchReceiver: TObject);
212 begin
213   Assert(Supports(CommDispatchReceiver, ICommDispatchReceiver, FCommDispatchReceiver), StrCommReceiverMustSupportInterface);
214 
215   inherited Create(TRUE);
216 
217   FCommThreadParams := TCommThreadParams.Create;
218 end;
219 
220 destructor TCommThread.Destroy;
221 begin
222   FCommDispatchReceiver.CommThreadTerminated(Self);
223 
224   FreeAndNil(FCommThreadParams);
225 
226   inherited;
227 end;
228 
229 function TCommThread.GetObject(const ObjectName: String): TObject;
230 begin
231   Result := FCommThreadParams.GetObject(ObjectName);
232 end;
233 
234 function TCommThread.GetParam(const ParamName: String): Variant;
235 begin
236   Result := FCommThreadParams.GetParam(ParamName);
237 end;
238 
239 procedure TCommThread.SendCommMessage(MessageId: Integer;
240   CommThreadParams: TCommThreadParams);
241 begin
242   FCommDispatchReceiver.QueueMessage(Self, MessageId, CommThreadParams);
243 end;
244 
245 procedure TCommThread.SendProgressMessage(ProgressID: Int64; Progress,
246   ProgressMax: Integer; AlwaysSend: Boolean);
247 begin
248   if (AlwaysSend) or (now > FNextSendTime) then
249   begin
250     // Send a status message to the comm receiver
251     SendCommMessage(CTID_PROGRESS, TCommThreadParams.Create
252       .SetParam(PRM_PROGRESS_ID, ProgressID)
253       .SetParam(PRM_PROGRESS, Progress)
254       .SetParam(PRM_PROGRESS_MAX, ProgressMax));
255 
256     if not AlwaysSend then
257       FNextSendTime := now + (FProgressFrequency * OneMillisecond);
258   end;
259 end;
260 
261 procedure TCommThread.SendStatusMessage(const StatusText: String;
262   StatusType: Integer);
263 begin
264   // Send a status message to the comm receiver
265   SendCommMessage(CTID_STATUS, TCommThreadParams.Create
266     .SetParam(PRM_STATUS_TEXT, StatusText)
267     .SetParam(PRM_STATUS_TYPE, StatusType));
268 end;
269 
270 function TCommThread.SetObject(const ObjectName: String;
271   Obj: TObject): TCommThread;
272 begin
273   Result := Self;
274 
275   FCommThreadParams.SetObject(ObjectName, Obj);
276 end;
277 
278 function TCommThread.SetParam(const ParamName: String;
279   ParamValue: Variant): TCommThread;
280 begin
281   Result := Self;
282 
283   FCommThreadParams.SetParam(ParamName, ParamValue);
284 end;
285 
286 
287 { TCommThreadDispatch }
288 
289 function TBaseCommThreadDispatch.Cancelled: Boolean;
290 begin
291   Result := State = ctsTerminating;
292 end;
293 
294 procedure TBaseCommThreadDispatch.CommThreadTerminated(Sender: TObject);
295 var
296   idx: Integer;
297 begin
298   FCSCommThreads.Enter;
299   try
300     Assert(Sender is TCommThread, StrSenderMustBeATCommThread);
301 
302     // Find the thread in the active thread list
303     idx := FActiveThreads.IndexOf(Sender);
304 
305     Assert(idx <> -1, StrUnableToFindTerminatedThread);
306 
307     // if we find it, remove it (we should always find it)
308     FActiveThreads.Delete(idx);
309   finally
310     FCSCommThreads.Leave;
311   end;
312 end;
313 
314 constructor TBaseCommThreadDispatch.Create(AOwner: TComponent);
315 begin
316   inherited;
317 
318   FCommThreadClass := TCommThread;
319 
320   FProcessQueueTimer := TTimer.Create(nil);
321   FProcessQueueTimer.Enabled := FALSE;
322   FProcessQueueTimer.Interval := 5;
323   FProcessQueueTimer.OnTimer := OnProcessQueueTimer;
324   FProgressFrequency := 200;
325 
326   FCommQueue := TCommQueue.Create;
327 
328   FActiveThreads := TList.Create;
329 
330   FCSReceiveMessage := TCriticalSection.Create;
331   FCSCommThreads := TCriticalSection.Create;
332 end;
333 
334 destructor TBaseCommThreadDispatch.Destroy;
335 begin
336   // Stop the queue timer
337   FProcessQueueTimer.Enabled := FALSE;
338 
339   TerminateActiveThreads;
340 
341   // Pump the queue while there are active threads
342   while CommThreadDispatchState <> ctsIdle do
343   begin
344     ProcessMessageQueue;
345 
346     sleep(10);
347   end;
348 
349   // Free everything
350   FreeAndNil(FProcessQueueTimer);
351   FreeAndNil(FCommQueue);
352   FreeAndNil(FCSReceiveMessage);
353   FreeAndNil(FCSCommThreads);
354   FreeAndNil(FActiveThreads);
355 
356   inherited;
357 end;
358 
359 procedure TBaseCommThreadDispatch.DoOnReceiveThreadMessage(Sender: TObject;
360   MessageId: Integer; CommThreadParams: TCommThreadParams);
361 begin
362   // Don't send the messages if we're being destroyed
363   if not (csDestroying in ComponentState) then
364   begin
365     if Assigned(FOnReceiveThreadMessage) then
366       FOnReceiveThreadMessage(Self, Sender, MessageId, CommThreadParams);
367   end;
368 end;
369 
370 procedure TBaseCommThreadDispatch.DoOnStateChange;
371 begin
372   if (Assigned(FOnStateChange)) and (not (csDestroying in ComponentState)) then
373     FOnStateChange(Self, FCommThreadDispatchState);
374 end;
375 
376 function TBaseCommThreadDispatch.GetActiveThreadCount: Integer;
377 begin
378   Result := FActiveThreads.Count;
379 end;
380 
381 function TBaseCommThreadDispatch.GetProcessQueueTimerInterval: Integer;
382 begin
383   Result := FProcessQueueTimer.Interval;
384 end;
385 
386 
387 function TBaseCommThreadDispatch.GetStateText: String;
388 begin
389   case State of
390     ctsIdle: Result := StrIdle;
391     ctsTerminating: Result := StrTerminating;
392     ctsActive: Result := StrActive;
393   end;
394 end;
395 
396 function TBaseCommThreadDispatch.NewThread(const ThreadName: String): TCommThread;
397 begin
398