时间戳

首页 > 解决方案 > 从/ on Redis Set并发读取/写入 - 单服务器多个客户端

c# - 从/ on Redis Set并发读取/写入 - 单服务器多个客户端

问题描述

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

假设队列包含这些项目：[(1,value1),(2,v2),(3,v3),(4,v4)]。我们想要的是密钥为 3 的项目仅被一个客户端查看，并发请求查看项目的密钥为 4 或任何其他密钥。

提前致谢。

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

标签： c#redisstackexchange.redis

解决方案

要使进程互斥，您可以使用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作为锁定资源的。这意味着在锁被释放或过期之前，任何其他进程都无法锁定资源。

至于实现 pub/sub 系统，我强烈建议您使用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);
    }
}

如果您需要以高速处理更大的消息，您也可以使用 aService Bus Queue而不是，这是两者之间的比较： https ://docs.microsoft.com/en-us/azure/service-bus-messaging /服务总线天蓝色和服务总线队列比较对比Azure Storage Queue

推荐阅读