2

问题是：我已经说过每个线程应该按顺序使用的 5 种方法

方法1（）；然后方法2（）；然后 Method3(); 然后 Method4(); 然后 Method5();

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

我想实现以下场景：

我希望线程 1 开始使用 method1 然后移动到 method2 [并行我希望线程 2 开始使用现在未使用的 method1]

然后当线程 1 移动到方法 3，线程 2 继续到方法 2 时，线程 3 应该开始使用现在空闲的方法 1，依此类推。

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);
        }
    }
4

1 回答 1

2

我将其写为答案，因为所需的说明不适合评论。

您似乎有一系列工作要做（在特定对象上，可能会或可能不会发生变异）。您还有许多线程来执行此管道。管道由 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);
        }
    }
}
于 2012-04-30T03:20:35.393 回答