c# - 如何强制线程使用 C# 中的方法组

Question

问题是:我说了每个线程应该依次使用的 5 个方法

方法1();然后是方法 2()；然后是方法 3()；然后是方法 4()；然后是 Method5();

我还有 5 个线程在运行，编号从 1 到 5

我想实现以下场景:

我希望线程 1 开始使用方法 1，然后转到方法 2 [并行地，我希望线程 2 开始使用现在未使用的方法 1]

然后，当线程一转到方法 3，线程二继续执行方法 2 时，线程 3 应该开始使用现在空闲的方法一，依此类推。

public void Execute(object OPCounter)
{
    //Method 1
    lock (thisLock)
    {
    FetchedInstructionQueue[PCounter] = Stager.Stage1(InstructionsMemory);
    }

    //Method 2
    lock (thisLock)
    {
    DecordedInstructionQueue[PCounter] = Stager.Stage2(FetchedInstructionQueue, regMem);
    }

    //Method 3
    lock (thisLock)
    {
    ALUResultQueue[PCounter] = Stager.Stage3(DecordedInstructionQueue);
    }

    lock (thisLock)
    {
    MemoryQueue[PCounter] = Stager.Stage4(DecordedInstructionQueue, memory, ALUResultQueue);
    }

    lock (thisLock)
    {
    object obj = Stager.Stage5(DecordedInstructionQueue, ALUResultQueue, regMem, memory, MemoryQueue);
    InternalWriter(PCounter, obj);
    }

}

///This is the initiator of threads

private void ExecuteBtn_Click(object sender, EventArgs e)
{
    InstructionsMemory = InstructionsTextBox.Text.Split('\n');
    FetchedInstructionQueue = new string[InstructionsMemory.Length];
    DecordedInstructionQueue = new Instruction[InstructionsMemory.Length];
    ALUResultQueue = new int[InstructionsMemory.Length];
    MemoryQueue = new int[InstructionsMemory.Length];
    Thread[] threads = new Thread[InstructionsMemory.Length];

    for (APCounter = 0; APCounter < InstructionsMemory.Length; APCounter = 5 + APCounter)
    {
        if (APCounter + 5 < InstructionsMemory.Length)
        {
            object s1 = APCounter;
            object s2 = APCounter + 1;
            object s3 = APCounter + 2;
            object s4 = APCounter + 3;
            object s5 = APCounter + 4;

            threads[APCounter] = new Thread(new ParameterizedThreadStart(Execute));
            threads[APCounter + 1] = new Thread(new ParameterizedThreadStart(Execute));
            threads[APCounter + 2] = new Thread(new ParameterizedThreadStart(Execute));
            threads[APCounter + 3] = new Thread(new ParameterizedThreadStart(Execute));
            threads[APCounter + 4] = new Thread(new ParameterizedThreadStart(Execute));

            threads[APCounter].Start(s1);
            threads[APCounter + 1].Start(s2);
            threads[APCounter + 2].Start(s3);
            threads[APCounter + 3].Start(s4);
            threads[APCounter + 4].Start(s5);
        }
    }

Answer 1

我将此作为答案写下来，因为所需的澄清不适合评论。

您似乎有一系列工作要完成(在特定对象上，它可能会发生变化，也可能不会发生变化)。您还有许多线程来执行此管道。流水线由 5 个阶段组成。

通常，对于管道，您希望管道中的每个步骤都有一个线程(即，第 1 步有一个线程，第 2 步有一个线程，第 3 步有一个线程，依此类推)。我们称此为选项 A。

您似乎想对其进行设置，以便线程跟随正在处理的对象。因此，线程 1 通过所有 5 个阶段覆盖对象 1，然后线程 2 覆盖对象 2，依此类推。目前尚不清楚您为什么要这样做，但无论如何让我们开始吧。我们称此为选项 B。

为简单起见，我将使用 3 个线程和 3 个阶段来展示选项。

选项 A:传统管道

3 个阶段，每个阶段 1 个线程，对象在阶段之间移动。

void Main()
{
    var stage1Queue = new BlockingCollection<object>(new ConcurrentQueue<object>());
    var stage2Queue = new BlockingCollection<object>(new ConcurrentQueue<object>());
    var stage3Queue = new BlockingCollection<object>(new ConcurrentQueue<object>());

    var threads = new Thread[] {new Thread(() => Stage1Worker(stage1Queue, stage2Queue)),
                                new Thread(() => Stage2Worker(stage2Queue, stage3Queue)),
                                new Thread(() => Stage3Worker(stage3Queue))
                               };

    foreach (var thread in threads) thread.Start();

    stage1Queue.Add("*");
    stage1Queue.Add("*");
    stage1Queue.Add("*");

    Console.ReadKey();
}

public void Stage1Worker(BlockingCollection<object> queue, BlockingCollection<object> next)
{
    foreach (var task in queue.GetConsumingEnumerable())
    {
        Console.WriteLine(task); // do work here, even mutating task if needed
        next.TryAdd(task.ToString() + "*"); // will always succeed for a ConcurrentQueue
    }
}

public void Stage2Worker(BlockingCollection<object> queue, BlockingCollection<object> next)
{
    foreach (var task in queue.GetConsumingEnumerable())
    {
        Console.WriteLine(task); // do work here, even mutating task if needed
        next.TryAdd(task.ToString() + "*"); // will always succeed for a ConcurrentQueue
    }
}

public void Stage3Worker(BlockingCollection<object> queue)
{
    foreach (var task in queue.GetConsumingEnumerable())
    {
        Console.WriteLine(task); // do work here, even mutating task if needed
        // no more work!
    }
}

选项 B:同步方法访问管道

这是一个很奇怪的问题，如果不知道其中的“原因”，就很难找到合适的解决方案。下面确保单个任务由单个线程执行，线程等待访问每个方法。但是，它并不能保证线程 1 执行任务 1，线程 2 执行任务 2 等。无论哪个线程准备就绪，都将执行“下一个”任务。

object stage1Lock = new object();
object stage2Lock = new object();
object stage3Lock = new object();

void Main()
{
    var tasks = new BlockingCollection<object>(new ConcurrentQueue<object>());

    var threads = new Thread[] {new Thread(() => Worker(1, tasks)),
                                new Thread(() => Worker(2, tasks)),
                                new Thread(() => Worker(3, tasks))
                               };

    foreach (var thread in threads) thread.Start();

    tasks.Add("*");
    tasks.Add("**");
    tasks.Add("***");
    tasks.Add("****");
    tasks.Add("*****");

    LINQPad.Util.ReadLine();
}

public void Worker(int id, BlockingCollection<object> tasks)
{
    foreach (var task in tasks.GetConsumingEnumerable())
    {   
        Console.WriteLine(id + " got task: " + task);

        lock (stage1Lock){
            Console.WriteLine(id + " - Stage 1: " + task);
        }

        lock (stage2Lock){
            Console.WriteLine(id + " - Stage 2: " + task);
        }

        lock (stage3Lock){
            Console.WriteLine(id + " - Stage 3: " + task);
        }
    }
}