我需要使用的HttpWebRequest为实现一个应用服务器进行并行请求何时实现节流机制(每秒请求数)。 我的C#应用程序必须发出每秒不超过80个请求发送到远程服务器。 该限制由远程服务管理员不作为硬性限制,但作为我的平台和他们之间的“SLA”强加的。
我怎样才能使用的HttpWebRequest时控制每秒请求数?
我需要使用的HttpWebRequest为实现一个应用服务器进行并行请求何时实现节流机制(每秒请求数)。 我的C#应用程序必须发出每秒不超过80个请求发送到远程服务器。 该限制由远程服务管理员不作为硬性限制,但作为我的平台和他们之间的“SLA”强加的。
我怎样才能使用的HttpWebRequest时控制每秒请求数?
我有同样的问题,所以我做了一个无法找到一个现成的解决方案,并在这里。 该想法是使用一个BlockingCollection<T>
添加需要处理和使用反应性扩展用率限制处理器订阅的项目。
油门类的重命名版本, 这个速率限制器
public static class BlockingCollectionExtensions
{
// TODO: devise a way to avoid problems if collection gets too big (produced faster than consumed)
public static IObservable<T> AsRateLimitedObservable<T>(this BlockingCollection<T> sequence, int items, TimeSpan timePeriod, CancellationToken producerToken)
{
Subject<T> subject = new Subject<T>();
// this is a dummyToken just so we can recreate the TokenSource
// which we will pass the proxy class so it can cancel the task
// on disposal
CancellationToken dummyToken = new CancellationToken();
CancellationTokenSource tokenSource = CancellationTokenSource.CreateLinkedTokenSource(producerToken, dummyToken);
var consumingTask = new Task(() =>
{
using (var throttle = new Throttle(items, timePeriod))
{
while (!sequence.IsCompleted)
{
try
{
T item = sequence.Take(producerToken);
throttle.WaitToProceed();
try
{
subject.OnNext(item);
}
catch (Exception ex)
{
subject.OnError(ex);
}
}
catch (OperationCanceledException)
{
break;
}
}
subject.OnCompleted();
}
}, TaskCreationOptions.LongRunning);
return new TaskAwareObservable<T>(subject, consumingTask, tokenSource);
}
private class TaskAwareObservable<T> : IObservable<T>, IDisposable
{
private readonly Task task;
private readonly Subject<T> subject;
private readonly CancellationTokenSource taskCancellationTokenSource;
public TaskAwareObservable(Subject<T> subject, Task task, CancellationTokenSource tokenSource)
{
this.task = task;
this.subject = subject;
this.taskCancellationTokenSource = tokenSource;
}
public IDisposable Subscribe(IObserver<T> observer)
{
var disposable = subject.Subscribe(observer);
if (task.Status == TaskStatus.Created)
task.Start();
return disposable;
}
public void Dispose()
{
// cancel consumption and wait task to finish
taskCancellationTokenSource.Cancel();
task.Wait();
// dispose tokenSource and task
taskCancellationTokenSource.Dispose();
task.Dispose();
// dispose subject
subject.Dispose();
}
}
}
单元测试:
class BlockCollectionExtensionsTest
{
[Fact]
public void AsRateLimitedObservable()
{
const int maxItems = 1; // fix this to 1 to ease testing
TimeSpan during = TimeSpan.FromSeconds(1);
// populate collection
int[] items = new[] { 1, 2, 3, 4 };
BlockingCollection<int> collection = new BlockingCollection<int>();
foreach (var i in items) collection.Add(i);
collection.CompleteAdding();
IObservable<int> observable = collection.AsRateLimitedObservable(maxItems, during, CancellationToken.None);
BlockingCollection<int> processedItems = new BlockingCollection<int>();
ManualResetEvent completed = new ManualResetEvent(false);
DateTime last = DateTime.UtcNow;
observable
// this is so we'll receive exceptions
.ObserveOn(new SynchronizationContext())
.Subscribe(item =>
{
if (item == 1)
last = DateTime.UtcNow;
else
{
TimeSpan diff = (DateTime.UtcNow - last);
last = DateTime.UtcNow;
Assert.InRange(diff.TotalMilliseconds,
during.TotalMilliseconds - 30,
during.TotalMilliseconds + 30);
}
processedItems.Add(item);
},
() => completed.Set()
);
completed.WaitOne();
Assert.Equal(items, processedItems, new CollectionEqualityComparer<int>());
}
}
油门()和样品()扩展方法(可观察)让您调节事件的快速序列变成了“慢”序列。
这里是一个博客帖子用一个例子的Sample(Timespan)
,以确保一个最大信号速率。
我原来的职位讨论了如何通过客户端行为扩展节流机制添加到WCF,但随后指出,我误解了问题(DOH!)。
总体来说,方法可以检查与确定,如果我们违反限速,不是一类。 还有的已经被很多的讨论围绕着如何检查速率违反。
节流方法调用在N秒中号请求
如果你违反了速率限制,然后睡一固定间隔,并再次检查。 如果没有,则继续前进,使HttpWebRequest的电话。