当前位置：首页 > article >正文

Redis实践之高级用法：管道、消息队列、事务、分布式锁

article 2024/11/18 4:21:43

一、概念与简介：

管道（Pipeline）的使用：
- 使用 CreateBatch 方法创建一个批处理对象。
- 将多个操作添加到批处理中，然后执行批处理以提高性能。
发布-订阅模式（Pub/Sub）的实现：
- 使用 GetSubscriber 方法获取订阅者对象。
- 使用 PublishAsync 方法发布消息。
- 使用 SubscribeAsync 方法订阅频道并处理接收到的消息。
事务（Transaction）的应用：
- 使用 CreateTransaction 方法创建一个事务对象。
- 将多个操作添加到事务中，然后执行事务以确保原子性。
分布式锁的实现：
- 使用 LockTakeAsync 方法尝试获取锁。
- 在获取锁后执行关键操作，并使用 LockReleaseAsync 方法释放锁。

二、高级用法实践

1、首先，在你的 ASP.NET Core 项目中安装 StackExchange.Redis 包：

dotnet add package StackExchange.Redis

2、在Program.cs文件，配置redis服务接口

//注册redis 服务连接接口IConnectionMultiplexer
builder.Services.AddSingleton<IConnectionMultiplexer>(c =>
{
    return ConnectionMultiplexer.Connect(builder.Configuration.GetConnectionString("Redis")??"localhost:6379");
});

3、appsetting.json文件，配置redis连接串

"ConnectionStrings": {
  "Redis": "localhost:6379"
},

4、新增RedisServices.cs文件，实现高级用法

public class RedisServices
{
    private readonly IConnectionMultiplexer _redis;
    public RedisServices(IConnectionMultiplexer redis)
    {
        _redis = redis;
    }
    /// <summary>
    /// redis 管道用法
    /// </summary>
    /// <returns></returns>
    public async Task UsePipelineAsync()
    {
        var db = _redis.GetDatabase(3);
        var batch = db.CreateBatch();

        var tasks = new List<Task>();
        for (int i = 0;i<100;i++)
        {
            _=batch.StringSetAsync($"key{i}", $"value{i}", TimeSpan.FromMinutes(1));
            //tasks.Add(batch.StringSetAsync($"key{i}",$"value{i}",TimeSpan.FromMinutes(1)));
        }
        batch.Execute();
        //await Task.WhenAll(tasks);
        await Console.Out.WriteLineAsync("Pipeline execution completed.");
    }
    /// <summary>
    /// redis 发布订阅用法：发布
    /// </summary>
    /// <param name="channel"></param>
    /// <param name="message"></param>
    /// <returns></returns>
    public async Task PublishAsync(RedisChannel channel,string message)
    {           
        var subscriber = _redis.GetSubscriber();
        await subscriber.PublishAsync(channel, message);
        Console.WriteLine($"Message '{message}' published to channel '{channel}'.");
    }
    //订阅
    public async Task SubscriberAsync(RedisChannel channel)
    {
        var subscriber = _redis.GetSubscriber();
        await subscriber.SubscribeAsync(channel, (ch,msg) => 
        {
            Console.WriteLine($"Received message '{msg}' from channel '{ch}'.");
        });
    }
    /// <summary>
    /// redis 事务用法
    /// </summary>
    /// <returns></returns>
    public async Task UseTransactionAsync()
    {
        var  db = _redis.GetDatabase(3);
        var trans = db.CreateTransaction();

        _=trans.StringSetAsync("key1", "new_value1");
        _=trans.StringSetAsync("keyk", "new_valuek");

        bool committed = await trans.ExecuteAsync();
        Console.WriteLine($"Transaction committed: {committed}");
    }
    /// <summary>
    /// redis 分布式锁用法
    /// </summary>
    /// <param name="lockkey"></param>
    /// <returns></returns>
    public async Task UseDistributedLockAsync(string lockkey)
    {
        var db = _redis.GetDatabase(3);
        string lockvalue = Guid.NewGuid().ToString();
        TimeSpan expiry = TimeSpan.FromMinutes(1);

        bool acquired = await db.LockTakeAsync(lockkey,lockvalue,expiry);
        if (acquired)
        {
            try
            {
                Console.WriteLine("Lock acquired, performing critical operation...");
                // Perform critical operation here
            }
            finally
            {
                await db.LockReleaseAsync(lockkey,lockvalue);
                Console.WriteLine("Lock released.");
            }
        }
        else
        {
            Console.WriteLine("Failed to acquire lock.");
        }
    }
    public async Task<bool> GetRedisLock()
    {
        var db = _redis.GetDatabase(3);
        string lockkey = "_redisLock_";
        string lockvalue = Guid.NewGuid().ToString();
        TimeSpan expiry = TimeSpan.FromMinutes(1);
        return await db.LockTakeAsync(lockkey, lockvalue, expiry);
    }
}

5、注册RedisServices

 //注册redis高级用法
 builder.RegisterType<RedisServices>();

6、新建RedisController.cs

[Route("api/[controller]")]
[ApiController]
public class RedisController : ControllerBase
{
    private readonly Repository.IRepository<Product> _repository;
    private readonly RedisServices _redisServices;
    public RedisController(Repository.IRepository<Product> repository,RedisServices redisServices)
    {
        _repository = repository;
        _redisServices = redisServices;
    }

    //测试redis高级用法
    [Route("UseRedis")]
    [HttpPost]
    public async Task<ActionResult> UseRedis()
    {
        //管道
        await _redisServices.UsePipelineAsync();

        //在 Redis 的发布/订阅（Pub/Sub）模式下，消息是即时传递的，这意味着消息只会发送给当前在线的订阅者。
        //如果在发布消息时没有任何订阅者，那么这些消息将会丢失。因此，如果你在发布消息之后才开始订阅，之前发布的消息将不会被接收到。
        RedisChannel channel = new("mychannel",RedisChannel.PatternMode.Literal);
        await _redisServices.SubscriberAsync(channel);//先订阅
        await _redisServices.PublishAsync(channel, "hello redis!");//再发布
        
        //事务
        await _redisServices.UseTransactionAsync();

        //分布式锁
        await _redisServices.UseDistributedLockAsync("distributeLock");

        for (int i = 0;i<5000; i++)
        {
            bool acquired = await _redisServices.GetRedisLock();
            if (acquired)
            {
                await Console.Out.WriteLineAsync($"第{i}次请求获取锁成功，并执行成功");
            }
            else
            {
                Thread.Sleep(1000);
            }
        }
        return Ok();
    }
}