用于限制每秒请求数的信号量不起作用

Question

我正在使用Google Analytics，该服务的并发请求限制为10个。我不得不以某种方式限制我的API，所以我决定使用信号量，但它似乎不起作用。同时触发所有请求。我在代码中找不到问题。

public async Task<SiteAnalyticsDTO> Handle(GetSiteAnalyticsParameter query)
{
var todayVisits = _googleAnalyticsService.GetTodayVisitsNumber();
var todayTraffic = _googleAnalyticsService.GetTodayTraffic();
var newAndReturningUsers = _googleAnalyticsService.GetNewAndReturningUsersNumber();
var averageSessionDuration = _googleAnalyticsService.GetAverageSessionDuration();
var deviceCategory = _googleAnalyticsService.GetSessionNumberByDeviceCategory();
var topPages = _googleAnalyticsService.GetTodaysTopPages();
var guestsAndRegisteredUsers = _googleAnalyticsService.GetGuestsVsRegisteredUsers();
var averageNumberOfSessionsPerDay = _googleAnalyticsService.GetAverageSessionsNumber();
var visitsPerWeekday = _googleAnalyticsService.GetTrafficByWeekday();
var visitsByHours = _googleAnalyticsService.GetTrafficByTimeOfDay();
var usersByPrefectures = _googleAnalyticsService.GetUsersByPrefectures();
var usersByCountry = _googleAnalyticsService.GetUsersByCountry();
var tasks = new List<Task>()
{
todayVisits, todayTraffic, newAndReturningUsers,
averageSessionDuration, deviceCategory, topPages,
guestsAndRegisteredUsers, averageNumberOfSessionsPerDay, visitsPerWeekday,
visitsByHours, usersByPrefectures, usersByCountry
};
var throttler = new SemaphoreSlim(MaxRequests, MaxRequests);
foreach(var task in tasks)
{
await throttler.WaitAsync();
try
{
await task;
await Task.Delay(1000); // It's important due to limits of Google Analytics requests (10 queries per second per IP address)
}
finally
{
throttler.Release();
}
}
await Task.WhenAll(tasks);
return new SiteAnalyticsDTO()
{
TodayVisits = await todayVisits,
TodayTraffic = await todayTraffic,
NewAndReturningUsers = await newAndReturningUsers,
AverageSessionDuration = await averageSessionDuration,
DeviceCategory = await deviceCategory,
TopPages = await topPages,
GuestsAndRegisteredUsers = await guestsAndRegisteredUsers,
AverageNumberOfSessionsPerDay = await averageNumberOfSessionsPerDay,
VisitsPerWeekday = await visitsPerWeekday,
VisitsByHours = await visitsByHours,
UsersByPrefectures = await usersByPrefectures,
UsersByCountry = await usersByCountry
};
}

这里有一些谷歌分析调用的示例方法：

public async Task<int> GetTodayVisitsNumber(List<long> listingIds = null)
{
string filter = GetFilter(listingIds);
var getReportsRequest = GetReportsRequestModel(GetTodayDateRange(), "ga:sessionCount", "ga:sessions", _configuration.MainViewId, filter);
var response = await _service.Reports.BatchGet(getReportsRequest).ExecuteAsync();
Console.WriteLine(response);
var data = response.Reports.FirstOrDefault();
return Convert.ToInt32(data?.Data.Totals[0].Values[0]);
}

Answer 1

所有请求都会同时触发。

让我们看看这里的

var todayVisits = _googleAnalyticsService.GetTodayVisitsNumber();
var todayTraffic = _googleAnalyticsService.GetTodayTraffic();
var newAndReturningUsers = _googleAnalyticsService.GetNewAndReturningUsersNumber();
var averageSessionDuration = _googleAnalyticsService.GetAverageSessionDuration();
var deviceCategory = _googleAnalyticsService.GetSessionNumberByDeviceCategory();
var topPages = _googleAnalyticsService.GetTodaysTopPages();
var guestsAndRegisteredUsers = _googleAnalyticsService.GetGuestsVsRegisteredUsers();
var averageNumberOfSessionsPerDay = _googleAnalyticsService.GetAverageSessionsNumber();
var visitsPerWeekday = _googleAnalyticsService.GetTrafficByWeekday();
var visitsByHours = _googleAnalyticsService.GetTrafficByTimeOfDay();
var usersByPrefectures = _googleAnalyticsService.GetUsersByPrefectures();
var usersByCountry = _googleAnalyticsService.GetUsersByCountry();

您正在存储这些方法中每一个的结果。当你使用括号标记如"；CCD_ 1"；调用该方法并将结果存储在CCD_ 2中。

然后将这些方法的结果存储在一个列表中，然后将每个方法的结果await与一个Semaphore一起存储，以限制一次可以等待的任务数。

问题是：每个await都会立即完成，因为在上面最初调用它们时，您已经(同步(等待了。

这会让您相信SemaphoreSlim不起作用，因为如果每个Task在等待时立即返回(因为它们已经被调用(，那么它们之间就没有时间了。

存储async方法以备以后使用，而不是一次调用所有方法
不能像var中那样存储委托，必须将其存储在显式类型的变量methodName();0中。

例如：

Func<Task<object>> todayVisits = _googleAnalyticsService.GetTodayVisitsNumber;

编辑们注意到，我不知道这些方法是如何返回I替换的对象的，以尽可能通用

现在，如果我们将每个变量存储在一个变量中，那将是非常麻烦的，所以与其将它们存储在单个变量中，不如直接将它们放在这样的列表中：

var awaitableTasks = new List<Func<Task<object>>>()
{
_googleAnalyticsService.GetTodayVisitsNumber,
_googleAnalyticsService.GetTodayTraffic,
_googleAnalyticsService.GetNewAndReturningUsersNumber,
_googleAnalyticsService.GetAverageSessionDuration,
_googleAnalyticsService.GetSessionNumberByDeviceCategory,
_googleAnalyticsService.GetTodaysTopPages,
_googleAnalyticsService.GetGuestsVsRegisteredUsers,
_googleAnalyticsService.GetAverageSessionsNumber,
_googleAnalyticsService.GetTrafficByWeekday,
_googleAnalyticsService.GetTrafficByTimeOfDay,
_googleAnalyticsService.GetUsersByPrefectures,
_googleAnalyticsService.GetUsersByCountry
};

因为这些新对象本身不是任务，而是返回Task的方法，所以我们必须更改存储和调用它们的方式，为此，我们将使用一个本地方法，因此我将检查我所做的每一个更改。

让我们创建Semaphore，并创建一个可以放置任务的地方来跟踪它们。

让我们还创建一个地方，当我们await这些任务时，我们可以存储它们的结果。

var throttler = new SemaphoreSlim(MaxRequests, MaxRequests);
var tasks = new List<Task>();
ConcurrentDictionary<string, object> results = new();

让我们创建一个具有几个职责的本地方法

1. 接受Func<Task<object>>作为参数
2. Await方法
3. 把那个方法的结果放在我们以后可以得到的地方
4. 即使遇到错误，也要释放Semphore

async Task Worker(Func<Task<object>> awaitableFunc)
{
try
{
resultDict.TryAdd(awaitableFunc.GetMethodInfo().Name, await awaitableFunc());
}
finally
{
throttler.Release();
}
}

编辑注意：您可以用lambda表达式实现同样的功能，但为了清晰和格式化，我更喜欢使用本地方法

启动工人并存储他们返回的任务。

那样。。如果在创建最后一对对象时还没有完成，那么我们可以等待它们完成，然后再创建最终对象(因为我们需要它们提供的所有结果来创建最终对象(。

foreach (var task in awaitableTasks)
{
await throttler.WaitAsync();
tasks.Add(Task.Run(() => Worker(task)));
}
// wait for the tasks to finish
await Task.WhenAll(tasks);

创建最终对象，然后返回。

return new SiteAnalyticsDTO()
{
TodayVisits = resultDict[nameof(_googleAnalyticsService.GetTodayVisitsNumber)],
TodayTraffic = resultDict[nameof(_googleAnalyticsService.GetTodayTraffic)],
NewAndReturningUsers = resultDict[nameof(_googleAnalyticsService.GetNewAndReturningUsersNumber)],
AverageSessionDuration = resultDict[nameof(_googleAnalyticsService.GetAverageSessionDuration)],
DeviceCategory = resultDict[nameof(_googleAnalyticsService.GetSessionNumberByDeviceCategory)],
TopPages = resultDict[nameof(_googleAnalyticsService.GetTodaysTopPages)],
GuestsAndRegisteredUsers = resultDict[nameof(_googleAnalyticsService.GetGuestsVsRegisteredUsers)],
AverageNumberOfSessionsPerDay = resultDict[nameof(_googleAnalyticsService.GetAverageSessionsNumber)],
VisitsPerWeekday = resultDict[nameof(_googleAnalyticsService.GetTrafficByWeekday)],
VisitsByHours = resultDict[nameof(_googleAnalyticsService.GetTrafficByTimeOfDay)],
UsersByPrefectures = resultDict[nameof(_googleAnalyticsService.GetUsersByPrefectures)],
UsersByCountry = resultDict[nameof(_googleAnalyticsService.GetUsersByCountry)]
};

把它们放在一起，我认为我们有一些可能有效的东西，或者至少可以很容易地修改以满足您的需求。

public static async Task<SiteAnalyticsDTO> Handle(GetSiteAnalyticsParameter query)
{
// store these methods so we can iterate and execute them later
var awaitableTasks = new List<Func<Task<object>>>()
{
_googleAnalyticsService.GetTodayVisitsNumber,
_googleAnalyticsService.GetTodayTraffic,
_googleAnalyticsService.GetNewAndReturningUsersNumber,
_googleAnalyticsService.GetAverageSessionDuration,
_googleAnalyticsService.GetSessionNumberByDeviceCategory,
_googleAnalyticsService.GetTodaysTopPages,
_googleAnalyticsService.GetGuestsVsRegisteredUsers,
_googleAnalyticsService.GetAverageSessionsNumber,
_googleAnalyticsService.GetTrafficByWeekday,
_googleAnalyticsService.GetTrafficByTimeOfDay,
_googleAnalyticsService.GetUsersByPrefectures,
_googleAnalyticsService.GetUsersByCountry
};
// create a way to limit the number of concurrent requests
var throttler = new SemaphoreSlim(MaxRequests, MaxRequests);
// create a place to store the tasks we create
var finalTasks = new List<Task>();
// make sure we have some where to put our results
ConcurrentDictionary<string, object> resultDict = new();
// make a worker that accepts one of those methods, invokes it
// then adds the result to the dict
async Task Worker(Func<Task<object>> awaitableFunc)
{
try
{
resultDict.TryAdd(awaitableFunc.GetMethodInfo().Name, await awaitableFunc());
}
finally
{
// make sure even if we encounter an error we still release the semphore
throttler.Release();
}
}
// iterate over the tasks, wait for the sempahore
// when we get a slot, create a worker and send it to the background
foreach (var task in awaitableTasks)
{
await throttler.WaitAsync();
finalTasks.Add(Task.Run(() => Worker(task)));
}
// wait for any remaining tasks to finish up in the background if they are still running
await Task.WhenAll(finalTasks);
// create the return object from the results of the dictionary
return new SiteAnalyticsDTO()
{
TodayVisits = resultDict[nameof(_googleAnalyticsService.GetTodayVisitsNumber)],
TodayTraffic = resultDict[nameof(_googleAnalyticsService.GetTodayTraffic)],
NewAndReturningUsers = resultDict[nameof(_googleAnalyticsService.GetNewAndReturningUsersNumber)],
AverageSessionDuration = resultDict[nameof(_googleAnalyticsService.GetAverageSessionDuration)],
DeviceCategory = resultDict[nameof(_googleAnalyticsService.GetSessionNumberByDeviceCategory)],
TopPages = resultDict[nameof(_googleAnalyticsService.GetTodaysTopPages)],
GuestsAndRegisteredUsers = resultDict[nameof(_googleAnalyticsService.GetGuestsVsRegisteredUsers)],
AverageNumberOfSessionsPerDay = resultDict[nameof(_googleAnalyticsService.GetAverageSessionsNumber)],
VisitsPerWeekday = resultDict[nameof(_googleAnalyticsService.GetTrafficByWeekday)],
VisitsByHours = resultDict[nameof(_googleAnalyticsService.GetTrafficByTimeOfDay)],
UsersByPrefectures = resultDict[nameof(_googleAnalyticsService.GetUsersByPrefectures)],
UsersByCountry = resultDict[nameof(_googleAnalyticsService.GetUsersByCountry)]
};
}

Answer 2

设置的问题是所有任务都在同一时间启动，只有它们的等待被抑制。限制等待没有任何作用。只有您的延续被延迟。目标服务批量接收所有请求。

我的建议是使用一个专用类来封装节流逻辑。似乎您需要限制并发性和发送请求的速率，而这些限制中的每一个都可以通过使用单独的SemaphoreSlim来实现。这里有一个简单的实现：

public class ThrottledExecution
{
private readonly SemaphoreSlim _concurrencySemaphore;
private readonly SemaphoreSlim _delaySemaphore;
private readonly TimeSpan _delay;
public ThrottledExecution(int concurrencyLimit, TimeSpan rateLimitTime,
int rateLimitCount)
{
// Arguments validation omitted
_concurrencySemaphore = new SemaphoreSlim(concurrencyLimit, concurrencyLimit);
_delaySemaphore = new SemaphoreSlim(rateLimitCount, rateLimitCount);
_delay = rateLimitTime;
}
public async Task<TResult> Run<TResult>(Func<Task<TResult>> action)
{
await _delaySemaphore.WaitAsync();
ScheduleDelaySemaphoreRelease();
await _concurrencySemaphore.WaitAsync();
try { return await action().ConfigureAwait(false); }
finally { _concurrencySemaphore.Release(); }
}
private async void ScheduleDelaySemaphoreRelease()
{
await Task.Delay(_delay).ConfigureAwait(false);
_delaySemaphore.Release();
}
}

以下是如何使用它：

public async Task<SiteAnalyticsDTO> Handle(GetSiteAnalyticsParameter query)
{
var throttler = new ThrottledExecution(MaxRequests, TimeSpan.FromSeconds(1), 1);
var todayVisits = throttler.Run(() => _service.GetTodayVisitsNumber());
var todayTraffic = throttler.Run(() => _service.GetTodayTraffic());
var newAndReturningUsers = throttler.Run(() => _service.GetNewAndReturningUsersNumber());
var averageSessionDuration = throttler.Run(() => _service.GetAverageSessionDuration());
var deviceCategory = throttler.Run(() => _service.GetSessionNumberByDeviceCategory());
var topPages = throttler.Run(() => _service.GetTodaysTopPages());
var guestsAndRegisteredUsers = throttler.Run(() => _service.GetGuestsVsRegisteredUsers());
var averageNumberOfSessionsPerDay = throttler.Run(() => _service.GetAverageSessionsNumber());
var visitsPerWeekday = throttler.Run(() => _service.GetTrafficByWeekday());
var visitsByHours = throttler.Run(() => _service.GetTrafficByTimeOfDay());
var usersByPrefectures = throttler.Run(() => _service.GetUsersByPrefectures());
var usersByCountry = throttler.Run(() => _service.GetUsersByCountry());
var tasks = new List<Task>()
{
todayVisits, todayTraffic, newAndReturningUsers,
averageSessionDuration, deviceCategory, topPages,
guestsAndRegisteredUsers, averageNumberOfSessionsPerDay, visitsPerWeekday,
visitsByHours, usersByPrefectures, usersByCountry
};
await Task.WhenAll(tasks);
return new SiteAnalyticsDTO()
{
TodayVisits = await todayVisits,
TodayTraffic = await todayTraffic,
NewAndReturningUsers = await newAndReturningUsers,
AverageSessionDuration = await averageSessionDuration,
DeviceCategory = await deviceCategory,
TopPages = await topPages,
GuestsAndRegisteredUsers = await guestsAndRegisteredUsers,
AverageNumberOfSessionsPerDay = await averageNumberOfSessionsPerDay,
VisitsPerWeekday = await visitsPerWeekday,
VisitsByHours = await visitsByHours,
UsersByPrefectures = await usersByPrefectures,
UsersByCountry = await usersByCountry,
};
}

部分成功的结果似乎对您没有用处，所以您可以考虑在ThrottledExecution类中添加一些自动取消逻辑。如果任务失败，则应取消所有挂起的和后续的异步操作。