CAP带你轻松玩转Asp.Net Core消息队列

2018-08-22 599

CAP是什么？

CAP是由我们园子里的杨晓东大神开发出来的一套分布式事务的决绝方案，是.Net Core Community中的第一个千星项目（目前已经1656 Star），具有轻量级、易使用、高性能等特点。

https://github.com/dotnetcore/CAP

本博客主要针对易用性这一点，展开叙述，一起看看CAP如何结合EF Core和RabbitMQ带领小白轻松走入分布式消息队列的世界。

准备

首先，你需要搭建一套RabbitMQ系统，搭建过程在此不再叙述，如果大家觉得麻烦，可以用我搭好的。

HostName: coderayu.cn UserName:guest Password:guest (仅仅可用作实验，数据丢失不负责)

创建Asp.Net Core 项目，并引入Nuget包

你可以运行以下下命令在你的项目中安装 CAP。

PM> Install-Package DotNetCore.CAP

如果你的消息队列使用的是 Kafka 的话，你可以：

PM> Install-Package DotNetCore.CAP.Kafka

如果你的消息队列使用的是 RabbitMQ 的话，你可以：

PM> Install-Package DotNetCore.CAP.RabbitMQ

CAP 提供了 Sql Server, MySql, PostgreSQL 的扩展作为数据库存储：

// 按需选择安装你正在使用的数据库
PM> Install-Package DotNetCore.CAP.SqlServer
PM> Install-Package DotNetCore.CAP.MySql
PM> Install-Package DotNetCore.CAP.PostgreSql

创建DbContext

因为我采用的是EF Core，所以首先要创建一个DbContext上下文，代码如下：

public class CapDbContext:DbContext
    {
        public CapDbContext(DbContextOptions options) : base(options)
        {
        }
    }

Startup配置

首先需要在ConfigureServices函数中进行相关服务的注入，对应的操作和功能解释如下：

public void ConfigureServices(IServiceCollection services)
        {
            //注入DbContext上下文，如果用的是Mysql可能还需要添加Pomelo.EntityFrameworkCore.MySql这个Nuget包
            services.AddDbContext<CapDbContext>(options =>
                options.UseMySql("Server=127.0.0.1;Database=testcap;UserId=root;Password=123456;"));

            //配置CAP
            services.AddCap(x =>
            {
                x.UseEntityFramework<CapDbContext>();

                //启用操作面板
                x.UseDashboard();
                //使用RabbitMQ
                x.UseRabbitMQ(rb =>
                {
                    //rabbitmq服务器地址
                    rb.HostName = "coderayu.cn";

                    rb.UserName = "guest";
                    rb.Password = "guest";

                    //指定Topic exchange名称，不指定的话会用默认的
                    rb.ExchangeName = "cap.text.exchange";
                });

                //设置处理成功的数据在数据库中保存的时间（秒），为保证系统新能，数据会定期清理。
                x.SucceedMessageExpiredAfter = 24 * 3600;

                //设置失败重试次数
                x.FailedRetryCount = 5;
            });

            services.AddMvc().SetCompatibilityVersion(CompatibilityVersion.Version_2_1);
        }

最后还要再Congiure中启用CAP中间件

public void Configure(IApplicationBuilder app, IHostingEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }
            //启用cap中间件
            app.UseCap();

            app.UseMvc();
        }

利用EF Core生成CAP数据库

再程序包管理控制台中依此输入以下命令行

PM> Add-Migration Init

PM> update-database

如果成成功执行，那么打开数据库，就可以看到用来存储CAP发送和接收数据的表格了。

表格中每列的含义如下：

消息的发送和订阅

我们直接在ValuesController的基础上进行改造。

在 Controller 中注入 ICapPublisher 然后使用 ICapPublisher 进行消息发送

private readonly ICapPublisher _publisher;

        public ValuesController(ICapPublisher publisher)
        {
            _publisher = publisher;
        }

发送消息

[HttpGet]
        public string Get(string message)
        {
            //"cap.test.queue"是发送的消息RouteKey，可以理解为消息管道的名称
            _publisher.Publish("cap.test.queue",message);

            return "发送成功";
        }

订阅消息

       //"cap.test.queue"为发送消息时的RauteKey，也可以模糊匹配
        //详情https://www.rabbitmq.com/tutorials/tutorial-four-dotnet.html
        [CapSubscribe("cap.test.queue")]
        public void HandleMessage(string message)
        {
            Console.Write(DateTime.Now.ToString()+"收到消息:"+message);
        }

Run

启动程序后，首先看到CAP启动成功

紧随其后，消费者也就是我们的订阅方法在RabbitMQ服务器上注册成功。

发送消息，发送成功，如下

发送后，立即在控制台看到了订阅方法输出的结果。

消息的失败重试

在订阅方法中，如果抛出异常，那么CAP就会认为该条消息处理失败，会自动进行重试，重试次数在前方已经进行了配置。

我们把订阅方法做一个改动，打印接收的信息到控制台中，并抛出异常

//"cap.test.queue"为发送消息时的RauteKey，也可以模糊匹配
        //详情https://www.rabbitmq.com/tutorials/tutorial-four-dotnet.html
        [CapSubscribe("cap.test.queue")]
        public void HandleMessage(string message)
        {
            Console.WriteLine(DateTime.Now.ToString()+"收到消息:"+message);
            throw new Exception("测试失败重试");
        }

可以看到，立即进行了三次重试

可是在前面，我们设置的失败重试次数是5次，为什么这里只重试三次吗？是不是要叫晓东过来改BUG了呢？当然不是。

观察发现，CAP重试的前三次是立即进行的，而后面的重试，是每隔一段时间进行的，当在分布式通讯的过程中，可能出现了问题确实不会立即修复解决，可能过了一定时间，系统就自动恢复了，如网络抖动。

CAP仪表盘

发送成功了五条消息，成功接收处理了三条，两条处理失败，处理失败的任务，我们可以直接在面板中进行重新消费，可谓非常方便。

同时，处理失败的消息，点击消息的编号后，可以查看到消息的内容和异常原因。

CAP如此强大，让消息队列这种高大上产品操作So Easy，学会了CAP,也可以吹牛说，我也懂分布式任务处理啦。

感谢晓东开发出如此强大的项目，同时感谢.Net Core Community。

参考 CAP Github wiki

https://github.com/dotnetcore/CAP/wiki

本博客Demo代码

https://github.com/liuzhenyulive/CAP.Demo

微信关注我们

原文链接：https://yq.aliyun.com/articles/628699

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

胶囊网络到底是什么东东？

胶囊网络(CapsNets)是一种可能会对深度学习产生深远影响的热门新型神经网络结构，特别是在计算机视觉领域。那么问题来了：计算机视觉相关问题不是已经解决了吗?我们不是都已经见过卷积神经网络(CNNs)在各种计算机视觉任务(如分类、定位、目标检测、语义分割或实例分割)中达到超人水平的例子吗?(如图1) 图1：为一些主要的计算机视觉任务。如今，这些任务都需要一个非常不同的卷积神经网络(CNN)架构，例如用于分类的残差网络（ResNet）、用于目标检测的目标检测系统(YOLO)、用于实例分割的Mask R-CNN等等。我们已经了解了卷积神经网络，但是:卷积神经网络要在大量的图像上进行训练(或者它们重复使用了神经网络的一部分)，而胶囊网络可以很好地推广使用更少的训练数据。卷积神经网络不能很好地处理模糊性，胶囊网络却可以，甚至可以在拥挤的场

2018-08-23

856

随着工作年限的增长，我们从一开始负责一个功能，再到负责一个模块的数据字典及框架设计。再到负责整个系统的需求评审及架构设计。这一路见证着程序猿的成长。但当我们逐步成为一名架构师，或是一名项目管理人员时，会发现一个项目的成功，会牵扯到各式各样的问题及风险。无论是系统本身要兼容快速发展的业务形态，还是由于人员因素导致的项目延迟，又或是系统代码的臃肿或是难以维护，亦是新人来后的一脸迷茫。那么下面，分享下，项目流程管理之我见。一、整体项目流程 1、需求评审与确认要求：PD会进行需求的整理并放入需求资源池。确定本期研发的功能需求，并开始需求评审，需求评审时，能够使技术人员能够完全理解本次需求的前因后果，作用，目标及整个流程。产出：该阶段主要为pd产出相关prd及demo，对需求进行宣讲，并记录疑问及难点。 2、模块流程文档要求：围绕着本次迭代的核心问题，编写整个模块的闭环业务流程。如有复杂逻辑，需要画出用例图、协作图等。同时，要给出该模块的非功能性需求，例如：调用量、日均增量、访问次数等待。产出：领域模型、开发模块架构图、技术架构图、人员分工（每个人负责哪个模...

2018-08-24

600

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。