快捷導(dǎo)航

.net使用cap實(shí)現(xiàn)消息異步處理

更新時(shí)間：2024年05月17日 10:05:15 作者：假裝我不帥

CAP 是一個(gè)基于 .NET Standard 的 C# 庫(kù),它是一種處理分布式事務(wù)的解決方案,同樣具有 EventBus 的功能,它具有輕量級(jí)、易使用、高性能等特點(diǎn),本文給大家介紹了.net下使用cap實(shí)現(xiàn)消息異步處理,需要的朋友可以參考下

介紹

CAP 是一個(gè)基于 .NET Standard 的 C# 庫(kù)，它是一種處理分布式事務(wù)的解決方案，同樣具有 EventBus 的功能，它具有輕量級(jí)、易使用、高性能等特點(diǎn)。

新建項(xiàng)目

新建.net7web項(xiàng)目

安裝依賴包

安裝軟件

安裝redis和Sql Server

修改代碼

新建RedisConfigModel

namespace CAPStu01.Models;

public class RedisConfigModel
{
    /// <summary>
    /// 服務(wù)器地址
    /// </summary>
    public string Host { get; set; }

    /// <summary>
    /// 端口號(hào)
    /// </summary>
    public int Port { get; set; }

    /// <summary>
    /// 密碼
    /// </summary>
    public string Pwd { get; set; }
}

修改appsettings.json

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.AspNetCore": "Warning"
    }
  },
  "ConnectionStrings": {
    "SQlServer": "server=127.0.0.1;User ID=sa;Password=xxxx;database=capstu;Encrypt=True;TrustServerCertificate=True;connection timeout=600;"
  },
  "RedisConfig": {
    "Host": "127.0.0.1",
    "Port": 6379,
    "Pwd": ""
  }
}

修改Program.cs

using CAPStu01.Models;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddControllers();
var redisConfig = builder.Configuration.GetSection("RedisConfig").Get<RedisConfigModel>();
var connectionStr = builder.Configuration.GetConnectionString("SQlServer") ?? "";
builder.Services.AddCap(x =>
{
    x.UseRedis(options =>
    {
        if (options.Configuration != null && redisConfig != null)
        {
            options.Configuration.EndPoints.Add(redisConfig.Host, redisConfig.Port);
            options.Configuration.DefaultDatabase = 0;
            options.Configuration.Password = redisConfig?.Pwd ?? "";
        }
    });
    x.UseSqlServer(sqlServerOptions =>
    {
        sqlServerOptions.Schema = "dbo";
        sqlServerOptions.ConnectionString = connectionStr;
    });
    //開啟面板
    x.UseDashboard(d =>
    {
        //允許匿名訪問
        d.AllowAnonymousExplicit = true;
    });
});
var app = builder.Build();

app.UseRouting();
app.MapControllers();
app.Run();

新建HomeController

using DotNetCore.CAP;
using Microsoft.AspNetCore.Mvc;

namespace CAPStu01.Controllers;

[ApiController]
public class HomeController:ControllerBase
{
    public HomeController()
    {
        
    }

    /// <summary>
    /// 發(fā)送消息
    /// </summary>
    /// <returns></returns>
    [HttpGet("/")]
    public IActionResult Index([FromServices]ICapPublisher capBus)
    {
        capBus.Publish("test.show.time","你好,CAP");
        return Content("發(fā)送消息成功");
    }
    
    /// <summary>
    /// 接受消息
    /// </summary>
    /// <param name="data"></param>
    [NonAction]
    [CapSubscribe("test.show.time")]
    public void ReceiveMessage(string data)
    {
        Console.WriteLine("message data is:" + data);
    }
}

結(jié)果

如果使用redis需要定期清理streams內(nèi)容

安裝freeredis，修改Program.cs

builder.Services.AddSingleton<IRedisClient>(new RedisClient($"{redisConfig.Host}:{redisConfig.Port},password={redisConfig.Pwd},defaultDatabase=0"));

新增清除方法

private readonly IRedisClient _redisClient;

public HomeController(IRedisClient redisClient)
{
    _redisClient = redisClient;
}

/// <summary>
/// 清除已處理的redis數(shù)據(jù)
/// </summary>
/// <returns></returns>
[HttpGet("/clear")]
public IActionResult ClearAckStream()
{
    var groups = _redisClient.XInfoGroups("test.show.time");
    var unreandMsgs = new List<string>();
    //獲取所有的未讀消息
    foreach (var group in groups)
    {
        if (group.pending > 0)
        {
            //有未讀消息
            var unReadList = _redisClient.XPending("test.show.time", group.name);
            if (unReadList.count > 0)
            {
                var groupInfo = _redisClient.XPending("test.show.time", group.name);
                var unreandList = _redisClient.XPending("test.show.time", group.name, groupInfo.minId, groupInfo.maxId,
                    groupInfo.count);
                foreach (var unre in unreandList)
                {
                    unreandMsgs.Add(unre.id);
                }
            }
        }
    }
    //獲取全部的消息
    var allMsgs = _redisClient.XRange("test.show.time", "-", "+");
    foreach (var msg in allMsgs)
    {
        if (unreandMsgs.Contains(msg.id))
        {
            //這個(gè)消息未讀則跳過
            continue;
        }
        //刪除已處理的消息
        _redisClient.XDel("test.show.time", msg.id);
    }

    return Content($"共處理未讀消息:{unreandMsgs.Count}個(gè)，已讀消息{allMsgs.Length}個(gè)");
}