相关疑难解决方法(0)

当我查看一些遗留应用程序代码时,我注意到它正在使用字符串对象来进行线程同步.我正在尝试解决此程序中的一些线程争用问题,并想知道这是否会导致如此奇怪的情况.有什么想法吗 ？

private static string mutex= "ABC";

internal static void Foo(Rpc rpc)
{
    lock (mutex)
    {
        //do something
    }
}

根据文件:

"a SemaphoreSlim不使用Windows内核信号量".

是否有任何特殊资源SemaphoreSlim使得Dispose在SemaphoreSlim不再使用遗嘱时呼叫很重要？

应用程序需要加载数据并缓存一段时间。我希望如果应用程序的多个部分想要同时访问同一个缓存键，缓存应该足够智能，只加载一次数据并将该调用的结果返回给所有调用者。然而，MemoryCache并不是这样做的。如果您并行访问缓存（这通常发生在应用程序中），它会为每次尝试获取缓存值创建一个任务。我认为这段代码会达到预期的结果，但事实并非如此。我希望缓存只运行一项GetDataAsync任务，等待它完成，然后使用结果来获取其他调用的值。

using Microsoft.Extensions.Caching.Memory;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace ConsoleApp4
{
    class Program
    {
        private const string Key = "1";
        private static int number = 0;

        static async Task Main(string[] args)
        {
            var memoryCache = new MemoryCache(new MemoryCacheOptions { });

            var tasks = new List<Task>();
            tasks.Add(memoryCache.GetOrCreateAsync(Key, (cacheEntry) => GetDataAsync()));
            tasks.Add(memoryCache.GetOrCreateAsync(Key, (cacheEntry) => GetDataAsync()));
            tasks.Add(memoryCache.GetOrCreateAsync(Key, (cacheEntry) => GetDataAsync()));

            await Task.WhenAll(tasks);

            Console.WriteLine($"The cached value was: {memoryCache.Get(Key)}");
        }

        public static async Task<int> GetDataAsync()
        {
            //Simulate getting a large …

介绍：

我正在构建一个单节点网络爬虫来简单地验证200 OK.NET Core 控制台应用程序中的URL 。我在不同的主机上有一组 URL，我用HttpClient. 我对使用 Polly 和 TPL Dataflow 还很陌生。

要求：

1. 我想支持与可配置的MaxDegreeOfParallelism.
2. 我想将任何给定主机的并行请求数限制为 1（或可配置）。这是为了429 TooManyRequests使用 Polly 策略优雅地处理每个主机的响应。或者，我可以使用断路器在收到一个429响应时取消对同一主机的并发请求，然后一次一个地处理该特定主机？
3. 我完全没有使用 TPL 数据流，而是使用 Polly Bulkhead 或其他一些机制来限制并行请求，但我不确定为了实现需求 #2 的配置会是什么样子。

当前实施：

我当前的实现是有效的，除了我经常看到x对同一主机的并行请求429大约在同一时间返回......然后，他们都暂停重试策略......然后，他们都猛烈抨击同一台主机再次同时经常仍然收到429s。即使我在整个队列中均匀分布同一主机的多个实例，我的 URL 集合也会因一些429最终仍开始生成s 的特定主机而超重。

收到 a 后429，我想我只想向该主机发送一个并发请求，以尊重远程主机并追求200s。

验证器方法：

public async Task<int> GetValidCount(IEnumerable<Uri> urls, CancellationToken cancellationToken)
{
    var validator = new TransformBlock<Uri, bool>(
        async u => (await _httpClient.GetAsync(u, HttpCompletionOption.ResponseHeadersRead, cancellationToken)).IsSuccessStatusCode, …

是否有类似于ConcurrentDictionary<TKey,TValue>和的数据结构Interlocked.Exchange<T>(...)，它允许您自动设置新值并检索分配给任意键的旧值？任何超载似乎都不可能AddOrUpdate(...)。

我有多个线程将数据写入公共源,并且我希望两个线程相互阻塞,当且仅当它们触及相同的数据时.

有办法专门锁定任意键会很好:

string id = GetNextId();
AquireLock(id);
try
{
    DoDangerousThing();
}
finally
{
    ReleaseLock(id);
}

如果没有其他人试图锁定相同的密钥,我希望他们能够同时运行.

我可以使用一个简单的互斥体字典来实现这一点,但是我需要担心驱逐旧的,未使用的锁,如果集合变得太大,这可能会成为一个问题.

是否存在此类锁定模式的现有实现.

我有一个Azure Service Bus队列,我收到1到10条带有相同"密钥"的消息.其中一条消息需要使用长时间运行的操作进行处理.完成后,数据库将被更新,其他消息将检查它.但是,与此同时,其他消息将被重新排队,以便不会丢失该过程.

但重点是这个长时间运行的操作不能同时运行同一个键,不应该多次运行.

这是我到目前为止所得到的:

void Main()
{
    Enumerable.Range(1, 1000)
              .AsParallel()
              .ForAll(async i => await ManageConcurrency(i % 2, async () => await Task.Delay(TimeSpan.FromSeconds(10)))); 
}

private readonly ConcurrentDictionary<int, SemaphoreSlim> _taskLocks = new ConcurrentDictionary<int, SemaphoreSlim>();

private async Task<bool> ManageConcurrency(int taskId, Func<Task> task)
{
    SemaphoreSlim taskLock = null;

    try
    {
        if (_taskLocks.TryGetValue(taskId, out taskLock))
        {
            if (taskLock.CurrentCount == 0)
            {
                Console.WriteLine($"{DateTime.Now.ToString("hh:mm:ss.ffffff")},  {taskId}, I found. No available.. Thread Id: {Thread.CurrentThread.ManagedThreadId}");
                return false;
            }

            taskLock.Wait();

            Console.WriteLine($"{DateTime.Now.ToString("hh:mm:ss.ffffff")},  {taskId}, I found and took. Thread Id: …

我需要一段代码，根据参数键只允许同时由 1 个线程执行：

    private static readonly ConcurrentDictionary<string, SemaphoreSlim> Semaphores = new();

    private async Task<TModel> GetValueWithBlockAsync<TModel>(string valueKey, Func<Task<TModel>> valueAction)
    {
        var semaphore = Semaphores.GetOrAdd(valueKey, s => new SemaphoreSlim(1, 1));

        try
        {
            await semaphore.WaitAsync();

            return await valueAction();
        }
        finally
        {
            semaphore.Release(); // Exception here - System.ObjectDisposedException
            if (semaphore.CurrentCount > 0 && Semaphores.TryRemove(valueKey, out semaphore))
            {
                semaphore?.Dispose();
            }
        }
    }

我时不时地收到错误：

The semaphore has been disposed. : System.ObjectDisposedException: The semaphore has been disposed.
   at System.Threading.SemaphoreSlim.CheckDispose()
   at System.Threading.SemaphoreSlim.Release(Int32 releaseCount)
   at Project.GetValueWithBlockAsync[TModel](String valueKey, Func`1 valueAction) …