.Net简单使用CAP+RabbitMQ完成事件的发布与订阅

简介

CAP 是一个EventBus&＃xff0c;同时也是一个在微服务或者SOA系统中解决分布式事务问题的一个框架【官网】

使用

• 第一步&＃xff1a;创建发布者(Angel.Cap.Publisher)与订阅者(Angel.Cap.Subscriber)两个应用程序&＃xff0c;在Nuget上引用主要依赖:DotNetCore.CAP、DotNetCore.CAP.RabbitMQ&＃xff1b;用的是Sql Server,还要引用后面三个引用
  
• 第二步&＃xff1a;搭建发布者 Angel.Cap.Publisher

1. 添加数据库连接字符串

"ConnectionStrings": { //连接字符串"AngelDBContext": "data source&＃61;.;initial catalog&＃61;AngelCapDB;User ID&＃61;sa;Password&＃61;sa;integrated security&＃61;True;MultipleActiveResultSets&＃61;True;App&＃61;EntityFramework"}


2. 添加DBContext

public class PubDBContext : DbContext{public PubDBContext(){}public PubDBContext(DbContextOptions<PubDBContext> options) : base(options){}protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder){base.OnConfiguring(optionsBuilder);}}


3. Configure和ConfigureServices使用

public void ConfigureServices(IServiceCollection services)
{services.AddControllersWithViews();//注入数据库services.AddDbContext<PubDBContext>(options &＃61;> options.UseSqlServer(Configuration["ConnectionStrings:AngelDBContext"]));//注册cap事件services.AddCap(x &＃61;>{x.UseEntityFramework<PubDBContext>();x.UseRabbitMQ(rb &＃61;>{rb.HostName &＃61; "localhost";rb.UserName &＃61; "admin";rb.Password &＃61; "admin";rb.Port &＃61; 5672;rb.VirtualHost &＃61; "Angel2022";rb.ExchangeName &＃61; "AngelCapExchange";rb.QueueMessageExpires &＃61; 24 * 3600 * 10; });});}public void Configure(IApplicationBuilder app, IWebHostEnvironment env, PubDBContext efCoreApp)
{if (env.IsDevelopment()){app.UseDeveloperExceptionPage();}else{app.UseExceptionHandler("/Home/Error");// The default HSTS value is 30 days. You may want to change this for production scenarios, see https://aka.ms/aspnetcore-hsts.app.UseHsts();}app.UseHttpsRedirection();app.UseStaticFiles();app.UseRouting();efCoreApp.Database.EnsureCreated();//数据库不存在自动创建app.UseAuthorization();app.UseEndpoints(endpoints &＃61;>{endpoints.MapControllerRoute(name: "default",pattern: "{controller&＃61;Home}/{action&＃61;Index}/{id?}");});
}


4. 创建发布接口

[Route("api/[controller]/[action]")]
[ApiController]
public class PubController : ControllerBase
{private ICapPublisher _capBus;public PubController(ICapPublisher capBus){this._capBus &＃61; capBus;}[HttpGet]public IActionResult SendMsg(){for (var i &＃61; 0; i < 1000; i&＃43;&＃43;){DateTime date &＃61; DateTime.Now;_capBus.Publish("angel", $"******发布者&＃xff0c;发布的内容为&＃xff1a; {date},i:{i}*****");}return Content("发送成功");}
}


第三步&＃xff1a;搭建订阅者Angel.Cap.Subscriber

1. 添加数据库连接字符串&＃xff0c;与发布者一样
2. 添加DBContext

public class SubDBContext : DbContext
{public SubDBContext(){}public SubDBContext(DbContextOptions<SubDBContext> options) : base(options){}protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder){base.OnConfiguring(optionsBuilder);}
}


3. Configure和ConfigureServices使用


public void ConfigureServices(IServiceCollection services)
{services.AddControllersWithViews();//注入数据库services.AddDbContext<SubDBContext>(options &＃61;> options.UseSqlServer(Configuration["ConnectionStrings:AngelDBContext"]));//注册cap事件services.AddCap(x &＃61;>{x.UseEntityFramework<SubDBContext>();x.UseRabbitMQ(rb &＃61;>{rb.HostName &＃61; "localhost";rb.UserName &＃61; "admin";rb.Password &＃61; "admin";rb.Port &＃61; 5672;rb.VirtualHost &＃61; "Angel2022";rb.ExchangeName &＃61; "AngelCapExchange";rb.QueueMessageExpires &＃61; 24 * 3600 * 10; //队列中消息自动删除时间(默认10天)});});}public void Configure(IApplicationBuilder app, IWebHostEnvironment env, SubDBContext efCoreApp)
{if (env.IsDevelopment()){app.UseDeveloperExceptionPage();}else{app.UseExceptionHandler("/Home/Error");// The default HSTS value is 30 days. You may want to change this for production scenarios, see https://aka.ms/aspnetcore-hsts.app.UseHsts();}app.UseHttpsRedirection();app.UseStaticFiles();app.UseRouting();efCoreApp.Database.EnsureCreated();//数据库不存在自动创建app.UseAuthorization();app.UseEndpoints(endpoints &＃61;>{endpoints.MapControllerRoute(name: "default",pattern: "{controller&＃61;Home}/{action&＃61;Index}/{id?}");});
}


4. 创建订阅者接口

[Route("api/[controller]/[action]")]
[ApiController]
public class SubController : ControllerBase
{private ILogger _log;public SubController(ILogger<SubController> log){this._log &＃61; log;}///

Cap 面板

在Angel.Cap.Subscriber上使用面板

1. 在ConfigureServices的AddCap上调用UseDashboard()

//注册cap事件
services.AddCap(x &＃61;>
{x.UseEntityFramework<SubDBContext>();x.UseRabbitMQ(rb &＃61;>{rb.HostName &＃61; "localhost";rb.UserName &＃61; "admin";rb.Password &＃61; "admin";rb.Port &＃61; 5672;rb.VirtualHost &＃61; "Angel2022";rb.ExchangeName &＃61; "Angel_RabbMQ_Exchange";rb.QueueMessageExpires &＃61; 24 * 3600 * 10; });x.UseDashboard();//使用Cap可视化面板
});


2. 浏览器打开&＃xff0c;域名后面加上cap
   https://localhost:44368/cap