c# - 使用 TPL 数据流的请求/响应模式
问题描述
在使用 TPL 数据流库时,我们需要一个请求/响应模式。我们的问题是我们有一个调用依赖服务的 .NET 核心 API。依赖服务限制并发请求。我们的 API 不限制并发请求;因此,我们一次可以收到数千个请求。在这种情况下,依赖服务将在达到其限制后拒绝请求。因此,我们实现了一个BufferBlock<T>
和一个TransformBlock<TIn, TOut>
. 性能稳定,效果很好。我们测试了我们的 API 前端,有 1000 个用户发出 100 个请求/秒,0 个问题。缓冲区块缓冲请求,并且转换块并行执行我们所需数量的请求。依赖服务接收我们的请求并做出响应。我们在转换块操作中返回该响应,一切都很好。我们的问题是缓冲区块和转换块断开连接,这意味着请求/响应不同步。我们遇到了一个请求将收到另一个请求者的响应的问题(请参阅下面的代码)。
具体到下面的代码,我们的问题出在GetContent
方法上。该方法是从我们 API 中的服务层调用的,该服务层最终是从我们的控制器调用的。下面的代码和服务层是单例的。SendAsync
缓冲区与转换块断开连接,以便ReceiveAsync
返回任意响应,而不一定是发出的请求。
所以,我们的问题是:有没有办法使用数据流块来关联请求/响应?最终目标是一个请求进入我们的 API,被发送到依赖的服务,然后返回给客户端。我们的数据流实现的代码如下。
public class HttpClientWrapper : IHttpClientManager
{
private readonly IConfiguration _configuration;
private readonly ITokenService _tokenService;
private HttpClient _client;
private BufferBlock<string> _bufferBlock;
private TransformBlock<string, JObject> _actionBlock;
public HttpClientWrapper(IConfiguration configuration, ITokenService tokenService)
{
_configuration = configuration;
_tokenService = tokenService;
_bufferBlock = new BufferBlock<string>();
var executionDataFlowBlockOptions = new ExecutionDataflowBlockOptions
{
MaxDegreeOfParallelism = 10
};
var dataFlowLinkOptions = new DataflowLinkOptions
{
PropagateCompletion = true
};
_actionBlock = new TransformBlock<string, JObject>(t => ProcessRequest(t),
executionDataFlowBlockOptions);
_bufferBlock.LinkTo(_actionBlock, dataFlowLinkOptions);
}
public void Connect()
{
_client = new HttpClient();
_client.DefaultRequestHeaders.Add("x-ms-client-application-name",
"ourappname");
}
public async Task<JObject> GetContent(string request)
{
await _bufferBlock.SendAsync(request);
var result = await _actionBlock.ReceiveAsync();
return result;
}
private async Task<JObject> ProcessRequest(string request)
{
if (_client == null)
{
Connect();
}
try
{
var accessToken = await _tokenService.GetTokenAsync(_configuration);
var httpRequestMessage = new HttpRequestMessage(HttpMethod.Post,
new Uri($"https://{_configuration.Uri}"));
// add the headers
httpRequestMessage.Headers.Add("Authorization", $"Bearer {accessToken}");
// add the request body
httpRequestMessage.Content = new StringContent(request, Encoding.UTF8,
"application/json");
var postRequest = await _client.SendAsync(httpRequestMessage);
var response = await postRequest.Content.ReadAsStringAsync();
return JsonConvert.DeserializeObject<JObject>(response);
}
catch (Exception ex)
{
// log error
return new JObject();
}
}
}
解决方案
您需要做的是用 id 标记每个传入的项目,以便您可以将数据输入与结果输出相关联。这是一个如何做到这一点的例子:
namespace ConcurrentFlows.DataflowJobs {
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
/// <summary>
/// A generic interface defining that:
/// for a specified input type => an awaitable result is produced.
/// </summary>
/// <typeparam name="TInput">The type of data to process.</typeparam>
/// <typeparam name="TOutput">The type of data the consumer expects back.</typeparam>
public interface IJobManager<TInput, TOutput> {
Task<TOutput> SubmitRequest(TInput data);
}
/// <summary>
/// A TPL-Dataflow based job manager.
/// </summary>
/// <typeparam name="TInput">The type of data to process.</typeparam>
/// <typeparam name="TOutput">The type of data the consumer expects back.</typeparam>
public class DataflowJobManager<TInput, TOutput> : IJobManager<TInput, TOutput> {
/// <summary>
/// It is anticipated that jobHandler is an injected
/// singleton instance of a Dataflow based 'calculator', though this implementation
/// does not depend on it being a singleton.
/// </summary>
/// <param name="jobHandler">A singleton Dataflow block through which all jobs are processed.</param>
public DataflowJobManager(IPropagatorBlock<KeyValuePair<Guid, TInput>, KeyValuePair<Guid, TOutput>> jobHandler) {
if (jobHandler == null) { throw new ArgumentException("Argument cannot be null.", "jobHandler"); }
this.JobHandler = JobHandler;
if (!alreadyLinked) {
JobHandler.LinkTo(ResultHandler, new DataflowLinkOptions() { PropagateCompletion = true });
alreadyLinked = true;
}
}
private static bool alreadyLinked = false;
/// <summary>
/// Submits the request to the JobHandler and asynchronously awaits the result.
/// </summary>
/// <param name="data">The input data to be processd.</param>
/// <returns></returns>
public async Task<TOutput> SubmitRequest(TInput data) {
var taggedData = TagInputData(data);
var job = CreateJob(taggedData);
Jobs.TryAdd(job.Key, job.Value);
await JobHandler.SendAsync(taggedData);
return await job.Value.Task;
}
private static ConcurrentDictionary<Guid, TaskCompletionSource<TOutput>> Jobs {
get;
} = new ConcurrentDictionary<Guid, TaskCompletionSource<TOutput>>();
private static ExecutionDataflowBlockOptions Options {
get;
} = GetResultHandlerOptions();
private static ITargetBlock<KeyValuePair<Guid, TOutput>> ResultHandler {
get;
} = CreateReplyHandler(Options);
private IPropagatorBlock<KeyValuePair<Guid, TInput>, KeyValuePair<Guid, TOutput>> JobHandler {
get;
}
private KeyValuePair<Guid, TInput> TagInputData(TInput data) {
var id = Guid.NewGuid();
return new KeyValuePair<Guid, TInput>(id, data);
}
private KeyValuePair<Guid, TaskCompletionSource<TOutput>> CreateJob(KeyValuePair<Guid, TInput> taggedData) {
var id = taggedData.Key;
var jobCompletionSource = new TaskCompletionSource<TOutput>();
return new KeyValuePair<Guid, TaskCompletionSource<TOutput>>(id, jobCompletionSource);
}
private static ExecutionDataflowBlockOptions GetResultHandlerOptions() {
return new ExecutionDataflowBlockOptions() {
MaxDegreeOfParallelism = Environment.ProcessorCount,
BoundedCapacity = 1000
};
}
private static ITargetBlock<KeyValuePair<Guid, TOutput>> CreateReplyHandler(ExecutionDataflowBlockOptions options) {
return new ActionBlock<KeyValuePair<Guid, TOutput>>((result) => {
RecieveOutput(result);
}, options);
}
private static void RecieveOutput(KeyValuePair<Guid, TOutput> result) {
var jobId = result.Key;
TaskCompletionSource<TOutput> jobCompletionSource;
if (!Jobs.TryRemove(jobId, out jobCompletionSource)) {
throw new InvalidOperationException($"The jobId: {jobId} was not found.");
}
var resultValue = result.Value;
jobCompletionSource.SetResult(resultValue);
}
}
}
另请参阅此答案以供参考。
推荐阅读
- c++ - 如何从源代码构建 glibcxx 3.4.21?
- cocoapods - 在 Swift 项目中包含 SQLCipher pod 时如何避免构建错误?
- python - Google 趋势 - 如果 isPartial 为真,则删除最后一行
- xquery - 如何在 xquery 中验证 ContextItem
- django - 为什么我的应用没有在 GKE 的浏览器中加载
- mysql - 在 Sakila DB 中检索演员 id 及其姓名及其地址
- javascript - Jest - 断言调用了一个模拟函数回调
- javascript - 反应:从选择中选择
- maven - org.openqa.selenium.WebDriverException;未知错误:Chrome 无法启动:异常退出 - 在 JENKINS 中执行测试套件时出错
- php - 每个表格的下载链接?