c# - 并发读取/写入 Redis 集 - 单服务器多客户端

Question

我们有一些应用程序在不同的 PC 上运行。必须有一个 Redis 服务器为所有应用程序保存一个共享的键/值队列。每个应用程序有2个线程，一个线程用于填充队列，另一个线程用于迭代和处理队列。

假设队列包含这些项目:[(1,value1),(2,v2),(3,v3),(4,v4)]。我们想要的是仅由一个客户端查看具有键 3 的项目和具有键 4 或任何其他键的并发请求查看项目。

• 使用 Redis 实现此目的的最佳方法是什么？
• 有没有办法通过字符串 SET 达到目标？
• 是否可以使用 Redis 实现发布/订阅系统？

提前致谢。

注意 用 C# 编写的客户端，StackExchange.Redis

Answer 1

要使进程互斥，您可以使用 RedLock.Net .这是一个 Distributed Lock Manager这就像一个 lock 语句，适用于无法相互了解的进程。这是一个例子:

public async Task ProcessMessage(Message message) 
{   
    // the thing we are trying to lock, i.e: "3"
    var resource = message.Key; 

    // determines how long will the lock be alive untill it's automatically released
    var expiry = TimeSpan.FromSeconds(30);

    // how long will the thread wait trying to acquire the lock
    var wait = TimeSpan.FromSeconds(10);

    // time span between each request to Redis trying to acquire the lock
    var retry = TimeSpan.FromSeconds(1);

    // blocks the thread until acquired or 'wait' timeout
    using (var redLock = await redlockFactory.CreateLockAsync(resource, expiry, wait, retry))
    {
        // make sure we got the lock
        if (redLock.IsAcquired)
        {
            // we successfully locked the resource, now other processes will have to wait
            ProcessMessageInternal(message.Value);
        }
        else 
        {
            // could't get the lock within the wait time
            // handle collision
        }
    }

    // the lock is automatically released at the end of the using block
    // which means the IDisposable.Dispose method makes a request to Redis to release the lock
}

请注意我是如何使用消息的 Key 作为锁定资源的。这意味着在锁被释放或过期之前，任何其他进程都无法锁定该资源。

关于实现发布/订阅系统，我强烈建议您使用 Azure Storage Queue , 创建一个 Queue Trigger并订阅您的程序。

这一切听起来很复杂，但实现起来却非常容易:您可以将应用程序的线程分成两个进程:

消息阅读器:每当消息到达时，他们只是将消息加入队列，如下所示:

// Retrieve storage account from connection string.
CloudStorageAccount storageAccount = CloudStorageAccount.Parse(
    CloudConfigurationManager.GetSetting("StorageConnectionString"));

// Create the queue client.
CloudQueueClient queueClient = storageAccount.CreateCloudQueueClient();

// Retrieve a reference to a queue.
CloudQueue queue = queueClient.GetQueueReference("myqueue");

// Create the queue if it doesn't already exist.
queue.CreateIfNotExists();

var message = // get message

var json = SerializeMessage(message);

// Create a message and add it to the queue.
CloudQueueMessage message = new CloudQueueMessage(json);
queue.AddMessage(message);

消息处理器:谁将通过使用 QueueTrigger 订阅队列，Visual Studio 有一个名为 Azure Functions 的项目模板 您只需要传递存储连接字符串和队列名称，它将为您处理并发。此过程将水平升级(意味着将有许多实例)，因此它需要与其兄弟互斥，并将通过使用 RedLock.Net 实现。 azure 函数将像这样锁定:

public class Functions 
{
    public static void ProcessQueueMessage([QueueTrigger(QueueName)] string serializedMessage)
    {
        var message = DeserializeMessage(serializedMessage);

        MessageProcesor.ProcessMessage(message);
    }
}

您还可以使用 Service Bus Queue如果您需要高速处理更大的消息，而不是 Azure 存储队列，下面是两者之间的比较:https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-azure-and-service-bus-queues-compared-contrasted