Update overview
- DelegateHandler
- Named RequestBatcher
- Observability
Project address:
NuGet：
This article is written based on the officially released RequestBatcher v0.3.4 source code and Chinese README:
/tree/core-v0.3.4
Preface
RequestBatcher is an in-process request batching library. The caller can submit requests independently. RequestBatcher merges the queued requests into batches, hands them to the application’s Handler for processing, and then returns the results to the corresponding caller respectively.
See the previous article for basic usage and processing semantics. This article introduces delegate Handler, Named RequestBatcher and observability.
In the existing memory benchmark, there is almost no difference in the stable calling performance of delegated Handler and Handler classes; this set of results does not include the compilation overhead of first binding. See the test conditions and original data:
/blob/core-v0.3.4/docs/evaluations/delegate-handlers/README.md
Delegate Handler: organize processing logic at the call site
Take batch query of product prices as an example. The original writing method needs to be implemented and registered. IRequestBatchHandler. in the example IPriceStore.FindAsync Receive the product ID and cancellation token and return the price dictionary. Missing products do not appear in the dictionary:
public sealed class PriceByIdHandler(
[FromKeyedServices("catalog")] IPriceStore store)
: IRequestBatchHandler
{
public async ValueTask HandleAsync(
IReadOnlyList> entries,
CancellationToken cancellationToken = default)
{
var ids = entries.Select(entry => entry.Request).Distinct().ToArray();
var pricesById = await store.FindAsync(ids, cancellationToken).ConfigureAwait(false);
entries.SetResponses(id =>
pricesById.TryGetValue(id, out var price) ? price : (decimal?)null);
}
}
Specify the Handler type when registering and pass handlerLifetime Set the life cycle of Handler:
services.AddRequestBatcher(
batcherName: "price",
handlerLifetime: ServiceLifetime.Scoped);
After using the delegate Handler,can be omitted PriceByIdHandler class, write the same query and backfill logic directly in ExecuteAsync call place：
using Microsoft.Extensions.DependencyInjection;
using RequestBatcher;
public sealed class PriceLookup(
[FromKeyedServices("price")] IRequestBatcher batcher)
{
public Task> FindAsync(
IEnumerable productIds,
CancellationToken cancellationToken = default) =>
batcher.ExecuteAsync(
productIds,
static async ValueTask (
IReadOnlyList> entries,
[FromKeyedServices("catalog")] IPriceStore store,
CancellationToken batchCancellation) =>
{
var ids = entries.Select(entry => entry.Request).Distinct().ToArray();
var pricesById = await store.FindAsync(ids, batchCancellation)
.ConfigureAwait(false);
entries.SetResponses(id =>
pricesById.TryGetValue(id, out var price) ? price : (decimal?)null);
},
cancellationToken);
}
Change the delegation method to AddDelegateRequestBatcher Registration, choose one of the above Handler class registration:
services.AddDelegateRequestBatcher(batcherName: "price");
services.AddScoped();
The delegation method does not need to be specified handlerLifetimethe processing delegate is bound by the RequestBatcher instance and reused across batches.
The services required by the delegate are resolved during each batch execution, still following the DI lifecycle of their respective registrations.
When submitting a request, pass ExecuteAsync of handler Parameters are passed into the processing delegate. It supports single request and explicit list submission, and also supports two forms: return value and no return value.
Parameter injection
The Handler class injects services through the constructor, and the delegate Handler injects services through method parameters.
RequestBatcher A service required to parse parameters from the DI container when executing a batch.
The above delegation parameters are:
entries: The request batch actually processed this time is provided by RequestBatcher and must be placed in the first parameter.store:pass[FromKeyedServices("catalog")]Inject the specified KeyIPriceStore; Ordinary service parameters do not need this attribute and are directly parsed by type.batchCancellation: The cancellation token for batch processing is provided by RequestBatcher. This parameter can be omitted.
The above Keyed service needs to be registered to the DI container:
services.AddKeyedScoped("catalog");
Here’s "catalog" Select a storage service,PriceLookup in the constructor "price" Select the RequestBatcher instance.
When using Lambda, you need to explicitly add staticcannot capture external local variables or instance state; required services are injected through delegate parameters.
The caller passes in the request, a processing delegate, and its own cancellation token; the delegate parameters are populated by the RequestBatcher.
When there are service parameters, an independent asynchronous DI Scope is created for each actual execution batch.
Scoped services are shared within batches, Singleton services are shared within containers, and Transient services are parsed separately for each parameter. The service comes from the batch’s own Scope. Even if a batch combines multiple HTTP requests, the Scope of one of the requests will not be used.
Processing delegates and callers are cached and service instances are resolved in batches. Before the caller receives the completion result, the asynchronous release of the batch Scope has been completed; when there are no service parameters, there is no need to create a Scope.
batchCancellation Belongs to the processing life cycle of RequestBatcher itself. Normal call DisposeAsync it will first stop receiving new requests, wait for received requests and batches in execution to complete, and finally cancel the token to stop the consumption cycle.
ExecuteAsync final cancellationToken Belongs to the caller and can only cancel its own request before distribution.
Once the request has been handed to the Handler, the actual processing results continue to be returned. Cancellation by a single caller will not abort the shared batch.
Static implementation and delegate binding
The incoming processing delegate must use a static implementation. In addition to the above static Lambda, it also supports static anonymous methods and explicit declarations as static local functions, and method groups that point to ordinary or generic static methods.
You can also choose the following static method writing method before the first submission to replace the previous Lambda implementation:
public sealed class PriceLookup(
[FromKeyedServices("price")] IRequestBatcher batcher)
{
public Task> FindAsync(
IEnumerable productIds,
CancellationToken cancellationToken = default) =>
batcher.ExecuteAsync(productIds, HandlePricesAsync, cancellationToken);
private static async ValueTask HandlePricesAsync(
IReadOnlyList> entries,
[FromKeyedServices("catalog")] IPriceStore store,
CancellationToken batchCancellation)
{
var ids = entries.Select(entry => entry.Request).Distinct().ToArray();
var pricesById = await store.FindAsync(ids, batchCancellation).ConfigureAwait(false);
entries.SetResponses(id =>
pricesById.TryGetValue(id, out var price) ? price : (decimal?)null);
}
}
The same RequestBatcher instance is only bound to a processing delegate once, and subsequent submissions need to pass in equal delegates.
Equality is judged based on the method and target object corresponding to the delegate, and there is no need to always pass in the same delegate object.
Repeated calls PriceLookup instance FindAsyncusing the same Lambda or the same static method. If an instance that has been bound to Lambda is passed to other method groups, an error will be reported before joining the queue.
Lambdas in different locations may correspond to different Handler targets even if the code is the same. When multiple businesses share a RequestBatcher instance, they can be unified through PriceLookup Submit, or use the same static method uniformly; different processing logic uses different batcherName。
Delegates can be passed via variables, fields, parameters, properties, or factories; each time ExecuteAsync All need to be passed in Handler.
Signatures and compile-time checks
Use in no return value mode AddDelegateRequestBatcher Registration, the first parameter of the delegation is IReadOnlyList。
Use return value mode IReadOnlyList，Each entry requires a set response, including one indicating not found null。
Delegate supports synchronization void、Task and ValueTask Return type, not supported async void、Task or ValueTask. Responses are backfilled with entries.
The parameter must be a required parameter and is not supported ref、out Wait for the reference parameter. Handler can omit the entire CancellationToken Parameters; at most one when declaring, and cannot be written = default。
precedent PriceLookup.FindAsync The caller token can have a default value, but the injection parameters of the delegate Handler cannot have a default value.
The NuGet package has two built-in Analyzer diagnostics that report compile errors by default:
RB0001: For the Handler that can determine the source, check whether it uses static implementation.RB0002: Check the delegate signature that can be determined, including the first parameter type, return type, parameter modifiers, default values and injected attributes, etc.
When a delegate is passed through a property, factory, or parameter, the Analyzer may not be able to determine the final target; such passes are still allowed, and the caller needs to ensure that the static implementation is actually used. The runtime continues to check the actual signature, service binding, and delegate consistency, making it impossible to reliably distinguish a static Lambda in the source code from a normal Lambda that does not capture variables.
Delegate automatically binds dependencies on dynamic code and reflection metadata,NativeAOT or cropping is not currently supported.
Named RequestBatcher
Both price query and inventory query receive product ID and can be registered separately as "price" and "stock". The following code replaces the previous separate registration:
services.AddDelegateRequestBatcher(
batcherName: "price",
configure: options =>
{
options.BatchSize = 100;
options.MaxConcurrency = 4;
});
services.AddDelegateRequestBatcher(
batcherName: "stock",
configure: options =>
{
options.BatchSize = 200;
options.MaxConcurrency = 2;
});
"price" return price,"stock" Returns the inventory quantity.
The two instances each hold their own Handler, queue, capacity and concurrency limit.
Parsing and Injection
A RequestBatcher instance with a specified name can be resolved directly from the application’s Service Provider:
var priceBatcher = serviceProvider
.GetRequiredKeyedService>("price");
var stockBatcher = serviceProvider
.GetRequiredKeyedService>("stock");
In the application service constructor, you can pass [FromKeyedServices("price")] Inject the corresponding RequestBatcher instance, the previous PriceLookup Just do it this way.
ASP.NET Core Controller Action and Minimal API Endpoint parameters also support injecting corresponding instances using the same attribute.
Unnamed registries continue to provide default services through ordinary GetRequiredService Or constructor injection parsing. Default instances and named instances can coexist.
Name, configuration and routing
The name is case-sensitive and cannot be nullempty string or blank.
Only one RequestBatcher instance can be registered with the same request type and name.even if the return type is different, the same name cannot be reused to register again.
For example, price query MaxConcurrency It is 4, and the inventory query is 2. When both access the same downstream, the total concurrency should be considered.
The parameters of each RequestBatcher are configured through registered callbacks and are fixed after registration. Runtime adjustment is not currently supported.
batcherName for selecting instances,UsePartitionKey Configure request routing within the instance. Replace the following code "price" Registration, routing by product ID:
services.AddDelegateRequestBatcher(
batcherName: "price",
configure: options =>
{
options.BatchSize = 100;
options.MaxConcurrency = 4;
options.UsePartitionKey(id => id, partitionCount: 8);
});
The same product ID enters the same Partition, and 8 Partitions share the 4 Handler execution quotas of the instance.
Requests with the same ID may still be divided into multiple batches that execute concurrently. Deduplication within the batch is completed by the Handler, and the execution order and cross-batch caching are controlled by the business.
Request merging still uses opportunistic batching: processing currently queued requests when there are execution quotas, without waiting for a fixed collection window. The same rules apply for queue capacity, exception callbacks, pre-dispatch cancellation, and drain requests on container release. RequestBatcher works within the process, and unfinished requests cannot be recovered after a process failure.
Observability
OpenTelemetry integration
core RequestBatcher The package is provided via the .NET built-in API namedRequestBatcher ActivitySource and Meter, do not rely on OpenTelemetry. OptionalRequestBatcher.OpenTelemetry The package provides registration extensions,
And decide whether to sample the Handler batch based on the sampling results submitted by the request. The two packages maintain version numbers independently.
Install the integration package, as well as the Hosting, Instrument and Exporter packages required by the application:
dotnet add package RequestBatcher
dotnet add package RequestBatcher.OpenTelemetry
dotnet add package OpenTelemetry.Extensions.Hosting
dotnet add package OpenTelemetry.Instrumentation.AspNetCore
dotnet add package OpenTelemetry.Exporter.OpenTelemetryProtocol
Register Tracing and Metrics in the application’s existing OpenTelemetry Provider:
using Microsoft.Extensions.DependencyInjection;
using OpenTelemetry.Metrics;
using OpenTelemetry.Trace;
builder.Services.AddOpenTelemetry()
.WithTracing(tracing => tracing
.AddAspNetCoreInstrumentation()
.AddRequestBatcherInstrumentation()
.AddOtlpExporter())
.WithMetrics(metrics => metrics
.AddRequestBatcherInstrumentation()
.AddOtlpExporter());
Simply installing the package will not automatically start collecting; you need to register the corresponding extension in the provider you are using.
link tracing
| Span | Follow content |
|---|---|
RequestBatcher.Execute |
once ExecuteAsync Submit, if there is the caller’s current Span, it will be used as its child Span, otherwise it will be the root Span. The time consuming includes capacity waiting, queuing, Handler processing and result completion. |
RequestBatcher.Handle |
A Handler batch is the root Span of an independent link; the database and HTTP calls collected through Instrument in the Handler are displayed as its child Span. |
When viewing the link, from Execute Jump along the Link to the actual processing of its request Handle. One submission can span multiple Handler batches.
A Handler batch can also merge multiple submitted requests. Link is used to jump between these links. It does not change the parent-child relationship of Span and does not guarantee the execution order.
Named RequestBatcher will append the registration name after the operation name, for example RequestBatcher.Execute pricing-queries andRequestBatcher.Handle pricing-queries. When filtering and viewing spans, you can focus on these properties:
requestbatcher.name: Configured Batcher name, omitted if not configured.requestbatcher.request.type: A short name that exposes the request type, e.g.Int32without namespace.requestbatcher.batch.size: The number of requests in a Handler batch.requestbatcher.batch.submission.count: How many different times has this Handler batch been merged?ExecuteAsyncsubmit.
Multiple requests submitted at the same time will only be counted once in each batch.
Handler does not inherit the caller’s HTTP Span, log Scope, Baggage or other AsyncLocal value.
Business data such as tenant identification should be passed explicitly through the request; the Activity created in the Handler will be normally associated with the current batch of Span.
sampling
After registering the Tracing extension, the application’s Sampler determines each Execute Span whether to sample.
The integration package corresponds to the request entering the current Handler batch. Execute Span, decide Handle Whether to sample:
| Execute Span corresponding to the current batch | Handle sampling behavior |
|---|---|
| At least one sampled | Sample this Handler batch. |
| All unsampled | This Handler batch is not sampled. |
The database and HTTP sub-Span within the Handler still use the application’s Sampler.ParentBasedSampler Indicates that the child Span follows the sampling decision of the parent Span,
The specified root sampler is used when there is no parent Span. For example ParentBasedSampler(new AlwaysOnSampler()) Will sample the root Span,
Child Span follows the parent. Integrated package versus standalone Handle Root Span uses the rules in the table above.
You can also choose not to install the integration package and register directly AddSource("RequestBatcher"). At this time, the applied Sampler will independently decide Handle
Whether to sample, so even if Execute has been sampled, the corresponding Handler processing link may still be missing. Limitations on the number of samples, exports, and links may also cause
The association is incomplete; when more associations need to be retained for large batches, the Span Link upper limit of the SDK should be adjusted. 0% sampling does not mean turning off Instrument completely.
For the implementation of Context and Link, as well as performance evaluation, see Observability Design.
Metrics
If you only need Metrics, you can choose not to install the integration package and monitor Meter directly:
builder.Services.AddOpenTelemetry()
.WithMetrics(metrics => metrics.AddMeter("RequestBatcher"));
Can also be installed RequestBatcher.OpenTelemetry，
Using the Metrics from the previous example AddRequestBatcherInstrumentation() Extension.
| Indicator name | type | unit | meaning |
|---|---|---|---|
requestbatcher.requests.submitted |
Counter |
{request}(ask) |
Number of requests received. |
requestbatcher.requests.processed |
Counter |
{request}(ask) |
The number of requests handled by the Handler, including failed processing. |
requestbatcher.batches.processed |
Counter |
{batch}(batch) |
The number of Handler batches processed. |
requestbatcher.batch.size |
Histogram |
{request}(ask) |
Number of requests per Handler batch. |
requestbatcher.batch.duration |
Histogram |
ms(millisecond) |
The Handler batch is time-consuming, including the Handler scope release. |
The dimension contains only the request type short name (requestbatcher.request.type) and optional Batcher name (requestbatcher.name）。
Types with the same name will share the request type dimension, and different Batcher names can be configured when differentiation is required. Request content, tenant values, or Partition Keys are not logged.
Indicator name already contains requestbatcher.;The sample Prometheus Exporter will generaterequestbatcher_batches_processed_total. Do not add repeatedly on the export side requestbatcher namespace/prefix,
To avoid appearing requestbatcher_requestbatcher_*。
self use System.Diagnostics.Metrics.MeterListener When collecting,You need to explicitly subscribe to the indicators you want to collect：
listener.InstrumentPublished = (instrument, meterListener) =>
{
if (instrument.Meter.Name == "RequestBatcher")
{
meterListener.EnableMeasurementEvents(instrument);
}
};
- Neither discovering a metric nor registering a measurement callback is equivalent to a subscription. Each listener is called once for each metric it wants to collect.
EnableMeasurementEvents；
If you want to collect only some indicators, you can pressinstrument.Namefilter. - Register for count and batch size
longCallback, time-consuming registrationdoublecallback. Called before submitting the requestStart()，
And let the listener survive until the Batcher is closed. Measurements taken before subscribing will not be reissued. - Callbacks are executed synchronously and may be called concurrently by multiple threads. They should be kept lightweight and thread-safe to avoid blocking I/O and exception propagation.
Tags that are still used after the callback returns need to be copied. Aggregation and export are implemented by custom Listener.
Logs
Built-in logging by applying existing Microsoft.Extensions.Logging Pipe output, no need to install OpenTelemetry integration package.
The log category prefix is RequestBatcher.RequestBatchCoordinator;Can be found in appsettings.json Chinese use RequestBatcher Prefix unified adjustment level:
{
"Logging": {
"LogLevel": {
"RequestBatcher": "Warning"
}
}
}
This configuration retains Warning and above levels; Debug can view the start and stop logs, and Trace will also record the start and completion of each batch.
The Handler failure log contains the original exception, the library itself does not log the request content.
Runnable example
The delegate Handler and Handler classes each have a separate set of PostgreSQL examples. The delegate version uses static Lambda at the submission point and injects Keyed in the batch Scope PriceStoreincluding price writing and POST /prices/query Batch query:
/tree/core-v0.3.4/samples/RequestBatcher.Delegates
The Handler class version uses a version named "price" The RequestBatcher instance directly receives the product ID and returns the nullable price:
/tree/core-v0.3.4/samples/RequestBatcher.Deduplication
In addition, the observability example starts Cart, Pricing, PostgreSQL and Grafana OTel LGTM through Docker Compose, and continuously generates requests to show the cross-service links and indicators after batching:
/tree/core-v0.3.4/samples/RequestBatcher.Observability
For complete parameter conventions and usage instructions, see the Chinese README:
/blob/core-v0.3.4/README.zh-CN.md
