c# - C# 中的线程和异步操作

Question

我是一只试图学习新技巧的老狗。我非常熟悉一种叫做 PowerBuilder 的语言，在这种语言中，当您想异步执行操作时，您会在新线程中生成一个对象。我要重申:整个对象在单独的线程中实例化并具有不同的执行上下文。该对象上的所有方法都在该单独线程的上下文中执行。

现在，我正在尝试使用 C# 实现一些异步执行，而 .NET 中的线程模型对我来说感觉完全不同。看起来我在一个线程中实例化对象，但我可以指定(在逐个调用的基础上)某些方法在不同的线程中执行。

差异看似微妙，但让我很沮丧。我的老派想法是，“我有一个 helper ，名叫鲍勃。鲍勃出去做事了。”如果我理解正确的话，新派思维是“我是 Bob。如果需要，我有时可以同时揉我的肚子和拍我的头。”

我的真实编码问题:我正在编写一个接口(interface)引擎，它通过 TCP 接收消息，将它们解析为可用数据，然后将该数据放入数据库。 “解析”一条消息大约需要一秒钟。根据解析的数据，数据库操作可能需要不到一秒，也可能需要十秒。 (所有时间都是为了澄清问题。)

我的老派想法告诉我，我的数据库类应该存在于一个单独的线程中，并且有类似 ConcurrentQueue 的东西。 .它只会在该队列上旋转，处理其中可能存在的任何内容。另一方面，解析器需要将消息推送到该队列中。这些消息将是(委托(delegate)？)诸如“根据此对象 中的数据创建订单”或“根据此对象 中的数据更新订单”之类的消息。可能值得注意的是，我实际上想以严格的单线程 FIFO 顺序处理“队列”中的“消息”。

基本上，我的数据库连接不能总是跟上我的解析器。我需要一种方法来确保我的解析器在我的数据库进程试图 catch 时不会减慢速度。有什么建议吗？

-- 编辑:用代码!每个人和每件事都告诉我使用 BlockingCollection。所以这里有一个简短的最终目标和代码的解释:

这将是一个 Windows 服务。启动时，它将产生多个“环境”，每个“环境”包含一个“dbworker”和一个“接口(interface)”。 “接口(interface)”将有一个“解析器”和一个“监听器”。

class cEnvironment {
    private cDBWorker MyDatabase;
    private cInterface MyInterface;

    public void OnStart () {
        MyDatabase = new cDBWorker ();
        MyInterface = new cInterface ();

        MyInterface.OrderReceived += this.InterfaceOrderReceivedEventHandler;

        MyDatabase.OnStart ();
        MyInterface.OnStart ();
    }

    public void OnStop () {
        MyInterface.OnStop ();
        MyDatabase.OnStop ();

        MyInterface.OrderReceived -= this.InterfaceOrderReceivedEventHandler;
    }

    void InterfaceOrderReceivedEventHandler (object sender, OrderReceivedEventArgs e) {
        MyDatabase.OrderQueue.Add (e.Order);
    }
}

class cDBWorker {
    public BlockingCollection<cOrder> OrderQueue = new BlockingCollection<cOrder> ();
    private Task ProcessingTask;

    public void OnStart () {
        ProcessingTask = Task.Factory.StartNew (() => Process (), TaskCreationOptions.LongRunning);
    }

    public void OnStop () {
        OrderQueue.CompleteAdding ();
        ProcessingTask.Wait ();
    }

    public void Process () {
        foreach (cOrder Order in OrderQueue.GetConsumingEnumerable ()) {
            switch (Order.OrderType) {
                case 1:
                    SuperFastMethod (Order);
                    break;

                case 2:
                    ReallySlowMethod (Order);
                    break;
            }
        }
    }

    public void SuperFastMethod (cOrder Order) {
    }

    public void ReallySlowMethod (cOrder Order) {
    }
}

class cInterface {
    protected cListener MyListener;
    protected cParser MyParser;

    public void OnStart () {
        MyListener = new cListener ();
        MyParser = new cParser ();

        MyListener.DataReceived += this.ListenerDataReceivedHandler;
        MyListener.OnStart ();
    }

    public void OnStop () {
        MyListener.OnStop ();
        MyListener.DataReceived -= this.ListenerDataReceivedHandler;
    }

    public event OrderReceivedEventHandler OrderReceived;

    protected virtual void OnOrderReceived (OrderReceivedEventArgs e) {
        if (OrderReceived != null)
            OrderReceived (this, e);
    }

    void ListenerDataReceivedHandler (object sender, DataReceivedEventArgs e) {
        foreach (string Message in MyParser.GetMessages (e.RawData)) {
            OnOrderReceived (new OrderReceivedEventArgs (MyParser.ParseMessage (Message)));
        }
    }

它编译。 (发货!)但这是否意味着我做对了？

Answer 1

BlockingCollection使将这种东西放在一起非常容易:

// the queue
private BlockingCollection<Message> MessagesQueue = new BlockingCollection<Message>();


// the consumer
private MessageParser()
{
    foreach (var msg in MessagesQueue.GetConsumingEnumerable())
    {
        var parsedMessage = ParseMessage(msg);
        // do something with the parsed message
    }
}

// In your main program
// start the consumer
var consumer = Task.Factory.StartNew(() => MessageParser(),
    TaskCreationOptions.LongRunning);

// the main loop
while (messageAvailable)
{
    var msg = GetMessageFromTcp();
    // add it to the queue
    MessagesQueue.Add(msg);
}

// done receiving messages
// tell the consumer that no more messages will be added
MessagesQueue.CompleteAdding();

// wait for consumer to finish
consumer.Wait();

消费者在队列上进行非忙碌等待，因此当没有可用资源时它不会占用 CPU 资源。