C# MSMQ 訊息佇列 Windows 服務架構：多執行緒處理、路由與組件調用完整實作

前言

Microsoft Message Queuing (MSMQ) 配合 .NET `System.Messaging` 命名空間，提供可靠的非同步訊息傳遞基礎設施。本文呈現一個完整的 **Windows 服務** 架構，支援多佇列監聽、多執行緒並行處理、三種路由模式（循環、屬性導向、組件調用）、錯誤佇列隔離與組件生命週期管理。

架構總覽

┌─────────────────────────────────────┐
│        Windows Service (ServiceControl)        │
│  ┌─────────────────────────────────┐  │
│  │  Hashtable     │  │  ← 每個 Input Queue 一個 CWorker
│  └─────────────────────────────────┘  │
└─────────────────────────────────────┘
                  │
        ┌─────────┼─────────┐
        ▼         ▼         ▼
   ┌────────┐ ┌────────┐ ┌────────┐
   │CWorker │ │CWorker │ │CWorker │  ← 對應 config 中的 ProcessDefinition
   └────┬───┘ └────┬───┘ └────┬───┘
        │          │          │
   ┌────┴───┐ ┌────┴───┐ ┌────┴───┐
   │Thread 1│ │Thread 1│ │Thread 1│  ← CWorkerThread 衍生類別
   │Thread 2│ │Thread 2│ │Thread 2│
   └────────┘ └────────┘ └────────┘

核心類別設計

1. 服務進入點：ServiceControl

using System.ServiceProcess;

public class ServiceControl : ServiceBase
{
    private Hashtable _workers = new Hashtable();

    public static void Main()
    {
        ServiceBase.Run(new ServiceControl());
    }

    public ServiceControl()
    {
        ServiceName = "MSDNMessageService";
        CanPauseAndContinue = true;
        AutoLog = false;
    }

    protected override void OnStart(string[] args)
    {
        LoadConfigurationAndCreateWorkers();
        foreach (CWorker w in _workers.Values) w.Start();
    }

    protected override void OnStop()
    {
        foreach (CWorker w in _workers.Values) w.Stop();
    }

    protected override void OnPause()
    {
        foreach (CWorker w in _workers.Values) w.Pause();
    }

    protected override void OnContinue()
    {
        foreach (CWorker w in _workers.Values) w.Continue();
    }

    private void LoadConfigurationAndCreateWorkers()
    {
        // 讀取 App.config 中的 ProcessList 區段
        IConfigCollection workers = ConfigManager.Get("ProcessList", new AppDomainSelector());
        foreach (IConfigItem item in workers)
        {
            var formatter = new WorkerFormatter
            {
                ProcessName = (string)item["ProcessName"],
                ProcessDesc = (string)item["ProcessDesc"],
                NumberThreads = (int)item["ProcessThreads"],
                InputQueue = (string)item["InputQueue"],
                ErrorQueue = (string)item["ErrorQueue"],
                ProcessType = ParseProcessType((int)item["ProcessType"])
            };
            
            // 讀取 OutputList 關聯輸出
            string query = $"SELECT * FROM OutputList WHERE ProcessName='{formatter.ProcessName}'";
            IConfigCollection outputs = ConfigManager.Get("OutputList", new ConfigQuery(query));
            formatter.OutputNames = outputs.Cast().Select(o => (string)o["OutputName"]).ToArray();

            _workers[formatter.ProcessName] = new CWorker(formatter);
        }
    }
}

2. 組態結構 (App.config)

<configuration>
  <configSections>
    <section name="ProcessList" type="System.Configuration.NameValueSectionHandler" />
    <section name="OutputList" type="System.Configuration.NameValueSectionHandler" />
  </configSections>

  <ProcessList>
    <ProcessDefinition 
        ProcessName="Worker1" 
        ProcessDesc="RoundRobin 2 Threads" 
        ProcessType="0" 
        ProcessThreads="2" 
        InputQueue=".\private$\input1" 
        ErrorQueue=".\private$\error" />
    <ProcessDefinition 
        ProcessName="Worker2" 
        ProcessDesc="Assembly 1 Thread" 
        ProcessType="2" 
        ProcessThreads="1" 
        InputQueue=".\private$\input2" 
        ErrorQueue=".\private$\error" />
  </ProcessList>

  <OutputList>
    <OutputDefinition ProcessName="Worker1" OutputName=".\private$\output1" />
    <OutputDefinition ProcessName="Worker2" OutputName="MyAssembly.dll" />
    <OutputDefinition ProcessName="Worker2" OutputName="MyNamespace.MyClass" />
  </OutputList>
</configuration>

3. 工作者基類：CWorker

public class CWorker
{
    private ArrayList _threads = new ArrayList();
    private WorkerFormatter _config;

    public CWorker(WorkerFormatter config) { _config = config; }

    public void Start()
    {
        for (int i = 0; i < _config.NumberThreads; i++)
        {
            CWorkerThread thread = CreateThread(_config.ProcessType);
            thread.Start(_config.InputQueue, _config.ErrorQueue, _config.OutputNames);
            _threads.Add(thread);
        }
    }

    public void Stop()   { foreach (CWorkerThread t in _threads) t.Stop(); }
    public void Pause()  { foreach (CWorkerThread t in _threads) t.Pause(); }
    public void Continue() { foreach (CWorkerThread t in _threads) t.Continue(); }

    private CWorkerThread CreateThread(int type)
    {
        return type switch
        {
            0 => new CWorkerThreadRoundRobin(),
            1 => new CWorkerThreadAppSpecific(),
            2 => new CWorkerThreadAssembly(),
            _ => throw new ArgumentException("Unknown type")
        };
    }
}

4. 執行緒抽象基類：CWorkerThread

public abstract class CWorkerThread
{
    protected MessageQueue _inputQueue;
    protected MessageQueue _errorQueue;
    protected string[] _outputNames;
    protected Thread _thread;
    protected bool _running;

    public virtual void Start(string inputPath, string errorPath, string[] outputs)
    {
        _inputQueue = new MessageQueue(inputPath);
        _errorQueue = new MessageQueue(errorPath);
        _outputNames = outputs;
        _running = true;
        _thread = new Thread(ProcessLoop) { IsBackground = true };
        _thread.Start();
    }

    public virtual void Stop()   { _running = false; _thread?.Join(5000); }
    public virtual void Pause()  { /* 暫停邏輯 */ }
    public virtual void Continue() { /* 繼續邏輯 */ }

    private void ProcessLoop()
    {
        while (_running)
        {
            try
            {
                // 接收訊息 (阻塞帶逾時)
                Message msg = _inputQueue.Receive(TimeSpan.FromSeconds(5));
                ProcessMessage(msg);
            }
            catch (MessageQueueException mqex) when (mqex.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)
            {
                continue; // 正常逾時，繼續迴圈
            }
            catch (Exception ex)
            {
                LogError(ex);
            }
        }
    }

    protected abstract void ProcessMessage(Message msg);
}

5. 三種處理模式實作

模式 1：循環分發 (RoundRobin)

public class CWorkerThreadRoundRobin : CWorkerThread
{
    private int _nextOutputIndex = 0;
    protected override void ProcessMessage(Message msg)
    {
        string output = _outputNames[_nextOutputIndex];
        _nextOutputIndex = (_nextOutputIndex + 1) % _outputNames.Length;
        SendToOutput(output, msg);
    }
}

模式 2：屬性導向 (AppSpecific)

public class CWorkerThreadAppSpecific : CWorkerThread
{
    protected override void ProcessMessage(Message msg)
    {
        int appSpecific = msg.AppSpecific; // 0-based index
        if (appSpecific >= 0 && appSpecific < _outputNames.Length)
            SendToOutput(_outputNames[appSpecific], msg);
        else
            SendToErrorQueue(msg, "Invalid AppSpecific");
    }
}

模式 3：組件方法調用 (Assembly)

// 介面契約
public interface IWebMessage
{
    WebMessageReturn Process(string label, string body, int appSpecific);
    void Release();
}

public enum WebMessageReturn { ReturnGood, ReturnBad, ReturnAbort }

public class CWorkerThreadAssembly : CWorkerThread
{
    private IWebMessage _component;

    public override void Start(string inputPath, string errorPath, string[] outputs)
    {
        base.Start(inputPath, errorPath, outputs);
        // outputs[0] = DLL 路徑, outputs[1] = 完整類別名
        Assembly asm = Assembly.LoadFrom(_outputNames[0]);
        _component = (IWebMessage)asm.CreateInstance(_outputNames[1]);
    }

    protected override void ProcessMessage(Message msg)
    {
        string body = msg.BodyStream.ReadToEndString(); // 需設定 Formatter
        var result = _component.Process(msg.Label, body, msg.AppSpecific);
        
        switch (result)
        {
            case WebMessageReturn.ReturnGood: break; // 成功，訊息已從佇列移除
            case WebMessageReturn.ReturnBad: SendToErrorQueue(msg, "Component returned Bad"); break;
            case WebMessageReturn.ReturnAbort: throw new Exception("Abort requested"); // 觸發回滾/重試
        }
    }

    public override void Stop()
    {
        _component?.Release(); // 釋放非受控資源 (DB 連線等)
        base.Stop();
    }
}

6. 錯誤處理與訊息格式化

private void SendToOutput(string path, Message msg)
{
    using var q = new MessageQueue(path);
    q.Send(msg); // 保留原訊息屬性
}

private void SendToErrorQueue(Message msg, string reason)
{
    using var q = new MessageQueue(_errorQueue.Path);
    msg.AppSpecific = -1; // 標記錯誤
    msg.Extension = System.Text.Encoding.UTF8.GetBytes(reason);
    q.Send(msg);
}

關鍵設計亮點

1. 配置驅動：新增佇列處理器無需改代碼，只修改 config。
2. 執行緒隔離：每佇列獨立執行緒池，互不干擾。
3. 組件熱插拔：`Assembly` 模式支援動態載入業務邏輯 DLL，`Release()` 確保資源釋放。
4. 錯誤隔離：失敗訊息導向錯誤佇列，不阻塞主流程，支援後續人工/自動重處理。
5. 服務控制語意：`Pause/Continue` 精確控制執行緒迴圈，配合 SCM 管理。

現代化遷移建議

• MSMQ 屬於舊技術，**新專案建議改用 RabbitMQ / Azure Service Bus / Amazon SQS / Kafka**。
• 遷移策略：定義共同介面 `IMessageProcessor`，保留 `CWorkerThread` 架構，僅替換底層收發實作。
• 容器化部署：MSMQ 需 Windows 容器且安裝 MSMQ 功能，跨平台困難。

總結

本架構展示了 .NET Framework 時代企業級訊息處理服務的完整骨幹：配置驅動、多執行緒、多模式路由、組件隔離、錯誤容錯。雖 MSMQ 漸趨淘汰，但**「工作者池 + 訊息泵 + 策略模式」**的架構思想在現代訊息系統（RabbitMQ Consumer、Azure Functions、Kubernetes Job）中依然普遍適用。