强制某些代码始终 运行 在同一个线程上
Forcing certain code to always run on the same thread
我们有一个旧的第 3 方系统（我们称之为 Junksoft® 95），我们通过 PowerShell 与之交互（它公开了一个 COM 对象），我正在将它包装在 REST 中 API（ASP.NET Framework 4.8 和 WebAPI 2）。我使用 System.Management.Automation nuget 包创建一个 PowerShell，在其中我将 Junksoft 的 COM API 实例化为一个 dynamic 对象，然后我使用：
//I'm omitting some exception handling and maintenance code for brevity
powerShell = System.Management.Automation.PowerShell.Create();
powerShell.AddScript("Add-Type -Path C:\Path\To\Junksoft\Scripting.dll");
powerShell.AddScript("New-Object Com.Junksoft.Scripting.ScriptingObject");
dynamic junksoftAPI = powerShell.Invoke()[0];
//Now we issue commands to junksoftAPI like this:
junksoftAPI.Login(user,pass);
int age = junksoftAPI.GetAgeByCustomerId(custId);
List<string> names = junksoftAPI.GetNames();
当我 运行 所有这些都在同一个线程上时（例如，在控制台应用程序中），这工作正常。但是，由于某些原因，当我将 junksoftAPI 放入 System.Web.Caching.Cache 并从我的网络应用程序中的不同控制器使用它时，这个 usually 不起作用。我说 通常 是因为当 ASP.NET 恰好将传入调用提供给创建 junksoftAPI 的线程时，这实际上有效。如果没有，Junksoft 95 会给我一个错误。
我有什么方法可以确保与 junksoftAPI 的所有交互都发生在 相同的 线程上吗？
注意我不想把整个web应用变成单线程应用！控制器和其他地方的逻辑应该像正常情况一样发生在不同的线程上。它应该只是发生在 Junksoft 特定线程上的 Junksoft 交互，如下所示：
[HttpGet]
public IHttpActionResult GetAge(...)
{
//finding customer ID in database...
...
int custAge = await Task.Run(() => {
//this should happen on the Junksoft-specific thread and not the next available thread
var cache = new System.Web.Caching.Cache();
var junksoftAPI = cache.Get(...); //This has previously been added to cache on the Junksoft-specific thread
return junksoftAPI.GetAgeByCustomerId(custId);
});
//prepare a response using custAge...
}
如果您没有说这是第 3 方工具，我会认为它是 GUI class。出于实际原因，让多个线程写入它们是一个非常糟糕的主意。 .NET 强制执行严格的 "only the creating thread shall write" 规则，从 2.0 onward.
一般的 WebServers，特别是 ASP.Net 使用相当大的线程池。我们说的是每核心 10 到 100 个线程。这意味着很难将任何请求确定为特定线程。你还不如不试试
同样，查看 GUI classes 可能是您最好的选择。您基本上可以制作一个单线程，其唯一目的是模仿 GUI 的事件队列。一般 Windows Forms 应用程序的 Main/UI 线程负责创建每个 GUI class 实例。它由 polling/processing 事件队列保持活动状态。它仅在通过事件队列接收到取消命令时结束。调度只是将订单放入该队列，因此我们可以避免跨线程问题。
您可以创建自己的单例工作线程来实现此目的。这是您可以将其插入您的 Web 应用程序的代码。
public class JunkSoftRunner
{
private static JunkSoftRunner _instance;
//singleton pattern to restrict all the actions to be executed on a single thread only.
public static JunkSoftRunner Instance => _instance ?? (_instance = new JunkSoftRunner());
private readonly SemaphoreSlim _semaphore;
private readonly AutoResetEvent _newTaskRunSignal;
private TaskCompletionSource<object> _taskCompletionSource;
private Func<object> _func;
private JunkSoftRunner()
{
_semaphore = new SemaphoreSlim(1, 1);
_newTaskRunSignal = new AutoResetEvent(false);
var contextThread = new Thread(ThreadLooper)
{
Priority = ThreadPriority.Highest
};
contextThread.Start();
}
private void ThreadLooper()
{
while (true)
{
//wait till the next task signal is received.
_newTaskRunSignal.WaitOne();
//next task execution signal is received.
try
{
//try execute the task and get the result
var result = _func.Invoke();
//task executed successfully, set the result
_taskCompletionSource.SetResult(result);
}
catch (Exception ex)
{
//task execution threw an exception, set the exception and continue with the looper
_taskCompletionSource.SetException(ex);
}
}
}
public async Task<TResult> Run<TResult>(Func<TResult> func, CancellationToken cancellationToken = default(CancellationToken))
{
//allows only one thread to run at a time.
await _semaphore.WaitAsync(cancellationToken);
//thread has acquired the semaphore and entered
try
{
//create new task completion source to wait for func to get executed on the context thread
_taskCompletionSource = new TaskCompletionSource<object>();
//set the function to be executed by the context thread
_func = () => func();
//signal the waiting context thread that it is time to execute the task
_newTaskRunSignal.Set();
//wait and return the result till the task execution is finished on the context/looper thread.
return (TResult)await _taskCompletionSource.Task;
}
finally
{
//release the semaphore to allow other threads to acquire it.
_semaphore.Release();
}
}
}
用于测试的控制台主要方法：
public class Program
{
//testing the junk soft runner
public static void Main()
{
//get the singleton instance
var softRunner = JunkSoftRunner.Instance;
//simulate web request on different threads
for (var i = 0; i < 10; i++)
{
var taskIndex = i;
//launch a web request on a new thread.
Task.Run(async () =>
{
Console.WriteLine($"Task{taskIndex} (ThreadID:'{Thread.CurrentThread.ManagedThreadId})' Launched");
return await softRunner.Run(() =>
{
Console.WriteLine($"->Task{taskIndex} Completed On '{Thread.CurrentThread.ManagedThreadId}' thread.");
return taskIndex;
});
});
}
}
}
输出：
请注意，虽然该函数是从不同线程启动的，但某些代码部分始终在 ID 为“5”的同一上下文线程上执行。
但请注意，尽管所有 Web 请求都是在独立线程上执行的，但它们最终会等待一些任务在单例工作线程上执行。这最终会在您的 Web 应用程序中造成瓶颈。这无论如何是你的设计限制。
以下是如何使用 BlockingCollection class:
从专用 STA 线程向 Junksoft API 发出命令
public class JunksoftSTA : IDisposable
{
private readonly BlockingCollection<Action<Lazy<dynamic>>> _pump;
private readonly Thread _thread;
public JunksoftSTA()
{
_pump = new BlockingCollection<Action<Lazy<dynamic>>>();
_thread = new Thread(() =>
{
var lazyApi = new Lazy<dynamic>(() =>
{
var powerShell = System.Management.Automation.PowerShell.Create();
powerShell.AddScript("Add-Type -Path C:\Path\To\Junksoft.dll");
powerShell.AddScript("New-Object Com.Junksoft.ScriptingObject");
dynamic junksoftAPI = powerShell.Invoke()[0];
return junksoftAPI;
});
foreach (var action in _pump.GetConsumingEnumerable())
{
action(lazyApi);
}
});
_thread.SetApartmentState(ApartmentState.STA);
_thread.IsBackground = true;
_thread.Start();
}
public Task<T> CallAsync<T>(Func<dynamic, T> function)
{
var tcs = new TaskCompletionSource<T>(
TaskCreationOptions.RunContinuationsAsynchronously);
_pump.Add(lazyApi =>
{
try
{
var result = function(lazyApi.Value);
tcs.SetResult(result);
}
catch (Exception ex)
{
tcs.SetException(ex);
}
});
return tcs.Task;
}
public Task CallAsync(Action<dynamic> action)
{
return CallAsync<object>(api => { action(api); return null; });
}
public void Dispose() => _pump.CompleteAdding();
public void Join() => _thread.Join();
}
使用 Lazy class 的目的是为了在构造动态对象期间通过将异常传播给调用者来显示可能的异常。
...exceptions are cached. That is, if the factory method throws an exception the first time a thread tries to access the Value property of the Lazy<T> object, the same exception is thrown on every subsequent attempt.
用法示例：
// A static field stored somewhere
public static readonly JunksoftSTA JunksoftStatic = new JunksoftSTA();
await JunksoftStatic.CallAsync(api => { api.Login("x", "y"); });
int age = await JunksoftStatic.CallAsync(api => api.GetAgeByCustomerId(custId));
如果您发现单个 STA 线程不足以及时处理所有请求，您可以添加更多 STA 线程，所有这些 运行 相同的代码（private readonly Thread[] _threads; ETC）。 BlockingCollection class 是线程安全的，可以被任意数量的线程同时使用。
我们有一个旧的第 3 方系统（我们称之为 Junksoft® 95），我们通过 PowerShell 与之交互（它公开了一个 COM 对象），我正在将它包装在 REST 中 API（ASP.NET Framework 4.8 和 WebAPI 2）。我使用 System.Management.Automation nuget 包创建一个 PowerShell，在其中我将 Junksoft 的 COM API 实例化为一个 dynamic 对象，然后我使用：
//I'm omitting some exception handling and maintenance code for brevity
powerShell = System.Management.Automation.PowerShell.Create();
powerShell.AddScript("Add-Type -Path C:\Path\To\Junksoft\Scripting.dll");
powerShell.AddScript("New-Object Com.Junksoft.Scripting.ScriptingObject");
dynamic junksoftAPI = powerShell.Invoke()[0];
//Now we issue commands to junksoftAPI like this:
junksoftAPI.Login(user,pass);
int age = junksoftAPI.GetAgeByCustomerId(custId);
List<string> names = junksoftAPI.GetNames();
当我 运行 所有这些都在同一个线程上时（例如，在控制台应用程序中），这工作正常。但是，由于某些原因，当我将 junksoftAPI 放入 System.Web.Caching.Cache 并从我的网络应用程序中的不同控制器使用它时，这个 usually 不起作用。我说 通常 是因为当 ASP.NET 恰好将传入调用提供给创建 junksoftAPI 的线程时，这实际上有效。如果没有，Junksoft 95 会给我一个错误。
我有什么方法可以确保与 junksoftAPI 的所有交互都发生在 相同的 线程上吗？
注意我不想把整个web应用变成单线程应用！控制器和其他地方的逻辑应该像正常情况一样发生在不同的线程上。它应该只是发生在 Junksoft 特定线程上的 Junksoft 交互，如下所示：
[HttpGet]
public IHttpActionResult GetAge(...)
{
//finding customer ID in database...
...
int custAge = await Task.Run(() => {
//this should happen on the Junksoft-specific thread and not the next available thread
var cache = new System.Web.Caching.Cache();
var junksoftAPI = cache.Get(...); //This has previously been added to cache on the Junksoft-specific thread
return junksoftAPI.GetAgeByCustomerId(custId);
});
//prepare a response using custAge...
}
如果您没有说这是第 3 方工具，我会认为它是 GUI class。出于实际原因，让多个线程写入它们是一个非常糟糕的主意。 .NET 强制执行严格的 "only the creating thread shall write" 规则，从 2.0 onward.
一般的 WebServers，特别是 ASP.Net 使用相当大的线程池。我们说的是每核心 10 到 100 个线程。这意味着很难将任何请求确定为特定线程。你还不如不试试
同样，查看 GUI classes 可能是您最好的选择。您基本上可以制作一个单线程，其唯一目的是模仿 GUI 的事件队列。一般 Windows Forms 应用程序的 Main/UI 线程负责创建每个 GUI class 实例。它由 polling/processing 事件队列保持活动状态。它仅在通过事件队列接收到取消命令时结束。调度只是将订单放入该队列，因此我们可以避免跨线程问题。
您可以创建自己的单例工作线程来实现此目的。这是您可以将其插入您的 Web 应用程序的代码。
public class JunkSoftRunner
{
private static JunkSoftRunner _instance;
//singleton pattern to restrict all the actions to be executed on a single thread only.
public static JunkSoftRunner Instance => _instance ?? (_instance = new JunkSoftRunner());
private readonly SemaphoreSlim _semaphore;
private readonly AutoResetEvent _newTaskRunSignal;
private TaskCompletionSource<object> _taskCompletionSource;
private Func<object> _func;
private JunkSoftRunner()
{
_semaphore = new SemaphoreSlim(1, 1);
_newTaskRunSignal = new AutoResetEvent(false);
var contextThread = new Thread(ThreadLooper)
{
Priority = ThreadPriority.Highest
};
contextThread.Start();
}
private void ThreadLooper()
{
while (true)
{
//wait till the next task signal is received.
_newTaskRunSignal.WaitOne();
//next task execution signal is received.
try
{
//try execute the task and get the result
var result = _func.Invoke();
//task executed successfully, set the result
_taskCompletionSource.SetResult(result);
}
catch (Exception ex)
{
//task execution threw an exception, set the exception and continue with the looper
_taskCompletionSource.SetException(ex);
}
}
}
public async Task<TResult> Run<TResult>(Func<TResult> func, CancellationToken cancellationToken = default(CancellationToken))
{
//allows only one thread to run at a time.
await _semaphore.WaitAsync(cancellationToken);
//thread has acquired the semaphore and entered
try
{
//create new task completion source to wait for func to get executed on the context thread
_taskCompletionSource = new TaskCompletionSource<object>();
//set the function to be executed by the context thread
_func = () => func();
//signal the waiting context thread that it is time to execute the task
_newTaskRunSignal.Set();
//wait and return the result till the task execution is finished on the context/looper thread.
return (TResult)await _taskCompletionSource.Task;
}
finally
{
//release the semaphore to allow other threads to acquire it.
_semaphore.Release();
}
}
}
用于测试的控制台主要方法：
public class Program
{
//testing the junk soft runner
public static void Main()
{
//get the singleton instance
var softRunner = JunkSoftRunner.Instance;
//simulate web request on different threads
for (var i = 0; i < 10; i++)
{
var taskIndex = i;
//launch a web request on a new thread.
Task.Run(async () =>
{
Console.WriteLine($"Task{taskIndex} (ThreadID:'{Thread.CurrentThread.ManagedThreadId})' Launched");
return await softRunner.Run(() =>
{
Console.WriteLine($"->Task{taskIndex} Completed On '{Thread.CurrentThread.ManagedThreadId}' thread.");
return taskIndex;
});
});
}
}
}
输出：
请注意，虽然该函数是从不同线程启动的，但某些代码部分始终在 ID 为“5”的同一上下文线程上执行。
但请注意，尽管所有 Web 请求都是在独立线程上执行的，但它们最终会等待一些任务在单例工作线程上执行。这最终会在您的 Web 应用程序中造成瓶颈。这无论如何是你的设计限制。
以下是如何使用 BlockingCollection class:
public class JunksoftSTA : IDisposable
{
private readonly BlockingCollection<Action<Lazy<dynamic>>> _pump;
private readonly Thread _thread;
public JunksoftSTA()
{
_pump = new BlockingCollection<Action<Lazy<dynamic>>>();
_thread = new Thread(() =>
{
var lazyApi = new Lazy<dynamic>(() =>
{
var powerShell = System.Management.Automation.PowerShell.Create();
powerShell.AddScript("Add-Type -Path C:\Path\To\Junksoft.dll");
powerShell.AddScript("New-Object Com.Junksoft.ScriptingObject");
dynamic junksoftAPI = powerShell.Invoke()[0];
return junksoftAPI;
});
foreach (var action in _pump.GetConsumingEnumerable())
{
action(lazyApi);
}
});
_thread.SetApartmentState(ApartmentState.STA);
_thread.IsBackground = true;
_thread.Start();
}
public Task<T> CallAsync<T>(Func<dynamic, T> function)
{
var tcs = new TaskCompletionSource<T>(
TaskCreationOptions.RunContinuationsAsynchronously);
_pump.Add(lazyApi =>
{
try
{
var result = function(lazyApi.Value);
tcs.SetResult(result);
}
catch (Exception ex)
{
tcs.SetException(ex);
}
});
return tcs.Task;
}
public Task CallAsync(Action<dynamic> action)
{
return CallAsync<object>(api => { action(api); return null; });
}
public void Dispose() => _pump.CompleteAdding();
public void Join() => _thread.Join();
}
使用 Lazy class 的目的是为了在构造动态对象期间通过将异常传播给调用者来显示可能的异常。
...exceptions are cached. That is, if the factory method throws an exception the first time a thread tries to access the
Valueproperty of theLazy<T>object, the same exception is thrown on every subsequent attempt.
用法示例：
// A static field stored somewhere
public static readonly JunksoftSTA JunksoftStatic = new JunksoftSTA();
await JunksoftStatic.CallAsync(api => { api.Login("x", "y"); });
int age = await JunksoftStatic.CallAsync(api => api.GetAgeByCustomerId(custId));
如果您发现单个 STA 线程不足以及时处理所有请求，您可以添加更多 STA 线程，所有这些 运行 相同的代码（private readonly Thread[] _threads; ETC）。 BlockingCollection class 是线程安全的，可以被任意数量的线程同时使用。