如果放弃PLINQ而仅显式使用Task Parallel Library，这将非常简单（尽管可能不那么简洁）：

// Limits the parallelism of the "expensive task"
var semaphore = new SemaphoreSlim(8);

var tasks = Enumerable.Range(1, 1000)
    .Select(x => Thread.CurrentThread.ManagedThreadId)
    .Select(async x =>
    {
        await semaphore.WaitAsync();
        var result = await Task.Run(() => Tuple.Create(x, Thread.CurrentThread.ManagedThreadId));
        semaphore.Release();

        return result;
    });

return Task.WhenAll(tasks).Result;

注意，我使用Tuple.Create来记录来自主线程的线程ID和来自派生任务的线程ID，从我的测试来看，前者对于每个元组总是相同的，而后者是变化的，这是应该的。
信号量确保并行度永远不会超过8（尽管创建元组的任务成本很低，但这不太可能）。如果达到8，任何新任务都将等待，直到信号量上有可用的位置。

您可以使用下面的OffloadQueryEnumeration方法，该方法可确保源序列的枚举将在枚举结果IEnumerable<TResult>的同一线程上进行。querySelector是一个委托，它将源序列的代理转换为ParallelQuery<T>。此查询在ThreadPool线程上内部枚举，但输出值将返回到当前线程。

/// <summary>
/// Enumerates the source sequence on the current thread, and enumerates
/// the projected query on a ThreadPool thread.
/// </summary>
public static IEnumerable<TResult> OffloadQueryEnumeration<TSource, TResult>(
    this IEnumerable<TSource> source,
    Func<IEnumerable<TSource>, IEnumerable<TResult>> querySelector)
{
    ArgumentNullException.ThrowIfNull(source);
    ArgumentNullException.ThrowIfNull(querySelector);
    object locker = new();
    (TSource Value, bool HasValue) input = default; bool inputCompleted = false;
    (TResult Value, bool HasValue) output = default; bool outputCompleted = false;
    using IEnumerator<TSource> sourceEnumerator = source.GetEnumerator();

    IEnumerable<TSource> GetSourceProxy()
    {
        while (true)
        {
            TSource sourceItem;
            lock (locker)
            {
                while (true)
                {
                    if (inputCompleted || outputCompleted) yield break;
                    if (input.HasValue) break;
                    Monitor.Wait(locker);
                }
                sourceItem = input.Value;
                input = default; Monitor.PulseAll(locker);
            }
            yield return sourceItem;
        }
    }

    IEnumerable<TResult> query = querySelector(GetSourceProxy());
    Task outputReaderTask = Task.Run(() =>
    {
        try
        {
            foreach (TResult result in query)
            {
                lock (locker)
                {
                    while (true)
                    {
                        if (outputCompleted) return;
                        if (!output.HasValue) break;
                        Monitor.Wait(locker);
                    }
                    output = (result, true); Monitor.PulseAll(locker);
                }
            }
        }
        finally
        {
            lock (locker) { outputCompleted = true; Monitor.PulseAll(locker); }
        }
    });

    // Main loop
    List<Exception> exceptions = new();
    while (true)
    {
        TResult resultItem;
        lock (locker)
        {
            // Inner loop
            while (true)
            {
                if (output.HasValue)
                {
                    resultItem = output.Value;
                    output = default; Monitor.PulseAll(locker);
                    goto yieldResult;
                }
                if (outputCompleted) goto exitMainLoop;
                if (!inputCompleted && !input.HasValue)
                {
                    // Fill the empty input slot, by reading the enumerator.
                    try
                    {
                        if (sourceEnumerator.MoveNext())
                            input = (sourceEnumerator.Current, true);
                        else
                            inputCompleted = true;
                    }
                    catch (Exception ex)
                    {
                        exceptions.Add(ex);
                        inputCompleted = true;
                    }
                    Monitor.PulseAll(locker); continue;
                }
                Monitor.Wait(locker);
            }
        }
    yieldResult:
        bool yieldOK = false;
        try { yield return resultItem; yieldOK = true; }
        finally
        {
            if (!yieldOK)
            {
                // The consumer stopped enumerating prematurely
                lock (locker) { outputCompleted = true; Monitor.PulseAll(locker); }
                Task.WhenAny(outputReaderTask).Wait();
            }
        }
    }
exitMainLoop:

    // Propagate possible exceptions
    try { outputReaderTask.GetAwaiter().GetResult(); }
    catch (OperationCanceledException) { throw; }
    catch (AggregateException aex) { exceptions.AddRange(aex.InnerExceptions); }

    if (exceptions.Count > 0)
        throw new AggregateException(exceptions);
}

此方法使用Monitor.Wait/Monitor.Pulse机制（教程），以便同步从一个线程到另一个线程的值传输。
用法示例：

int[] threadIds = Enumerable
    .Range(1, 1000)
    .Select(x => Thread.CurrentThread.ManagedThreadId)
    .OffloadQueryEnumeration(proxy => proxy
        .AsParallel()
        .AsOrdered()
        .WithDegreeOfParallelism(8)
        .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
        .Select(x => x)
    )
    .ToArray();

Online demo.
OffloadQueryEnumeration是一个非常复杂的方法，它不停地处理三个线程：
1.既枚举source序列又使用PLINQ生成的元素的当前线程，在这两个操作之间交替。
1.枚举PLINQ生成的序列的ThreadPool线程（outputReaderTask）。
1.由PLINQ机制执行任务以从GetSourceProxy()迭代器获取下一项的工作线程。此线程并非始终相同，但在任何给定时刻，最多只有一个工作线程被分配此任务。
所以很多事情都在发生，隐藏的bug有很多机会不被发现地通过，这是一种需要编写十几个测试的API，来Assert许多可能的场景的正确性（例如源序列中的故障、PLINQ操作符中的故障、使用者中的故障、取消、放弃枚举等）。我已经手动测试了其中的一些场景，但是我还没有编写任何测试，所以使用这个方法要小心。