事件总线--课件

Posted qingyunye

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了事件总线--课件相关的知识,希望对你有一定的参考价值。

事件总线

什么是事物

例如:事物 所有看到的一切都是事物,不能看到的也是事物

例如:团队微服务,成员微服务,聚合微服务,网关api,认证中心等等包括类,对象

所有的事件都是事物变化的结果

大家接触事件最早就是在js 或者是c#高级特性。大家对于事件不默认,但是对于事件不是很好理解

什么是事件

事件就是指事物状态的变化,每一次事物变化的结果都称作为事件

什么是事件总线

就是用来管理所有的事件的一种机制就称作为事件总线

包括事件发布,事件存储,事件订阅,事件处理的统称

作用:

事件总线是一种机制,它允许不同的组件彼此通信而不彼此了解。 组件可以将事件发送到Eventbus,而无需知道是谁来接听或有多少其他人来接听。 组件也可以侦听Eventbus上的事件,而无需知道谁发送了事件。 这样,组件可以相互通信而无需相互依赖。 同样,很容易替换一个组件。 只要新组件了解正在发送和接收的事件,其他组件就永远不会知道.

为什么要使用事件总线

将微服务系统各组件之间进行解耦

使用业务的发展来说

事件总线框架

CAP

masstransit

CAP内部概念

事件 : 就是一些状态信息

发布者:发布事件的角色 cap

订阅者:订阅消费事件的角色 cap

消息传输器:传输事件

消息存储器:存储事件

CAP存储事件消息队列类型Transport

Azure

rabbitmq

kafaka

In Memory Queue

CAP存储事件持久化类型

SQL Server

mysql

PostgreSQL

MongoDB

InMemoryStorage

CAP事件监控

Dashboard

微服务系统中如何使用CAP

条件

1、微服务系统

2、RabbitMQ

3、SqlServer

4、CAP

步骤

1、微服务系统准备

微服务系统全部准备完毕

2、RabbitMQ准备

2.1 环境准备

Erlang下载地址:https://www.erlang.org/downloads

RabbitMQ下载地址:https://www.rabbitmq.com/download.html

2.2 RabbitMQ 启动

1、在安装目录下添加可视化插件

rabbitmq-plugins enable rabbitmq_management

2、在安装目录下启动

    rabbitmq-server 

3、查看rabbitmq状态

    rabbitmqctl status

4、在浏览器输入http://127.0.0.1:15672

访问rabbitmq后台系统

3、SqlServer准备

SqlServer启动,安装

4、CAP准备

4.1 CAP环境

CAP官网地址:https://cap.dotnetcore.xyz/user-guide/zh/monitoring/dashboard/

4.2 CPA配置

1、在RuanMou.MicroService.Core项目中添加依赖

CAP Nuget DotNetCore.CAP
CAP传输器Nuget DotNetCore.CAP.RabbitMQ
CAP持久化DotNetCore.CAP.SqlServer

2、在RuanMou.MicroService.AggregateService服务中startup.cs中添加

 // 8、添加事件总线cap
          services.AddCap(x => {
              // 8.1 使用内存存储消息(消息发送失败处理)
              x.UseInMemoryStorage();
?
              // 8.2 使用RabbitMQ进行事件中心处理
              x.UseRabbitMQ(rb => {
                  rb.HostName = "localhost";
                  rb.UserName = "guest";
                  rb.Password = "guest";
                  rb.Port = 5672;
                  rb.VirtualHost = "/";
              });
          });

2.1 在AggregateController.cs中注入ICapPublisher

    private readonly ICapPublisher capPublisher;
public TeamsController(ICapPublisher capPublisher)
  {
      this.capPublisher = capPublisher;
  }

3、在RuanMou.MicroService.VideoService服务startup.cs中添加

 // 8、添加事件总线cap
          services.AddCap(x => {
              // 8.1 使用RabbitMQ进行事件中心处理
              x.UseRabbitMQ(rb => {
                  rb.HostName = "localhost";
                  rb.UserName = "guest";
                  rb.Password = "guest";
                  rb.Port = 5672;
                  rb.VirtualHost = "/";
              });
          });

3.1 在VideoController.cs 中方法上添加特性[CapSubscribe]

       /// <summary>
      /// 视频添加(异步添加)
      /// </summary>
      /// <param name="Video"></param>
      /// <returns></returns>
      [NonAction]
      [CapSubscribe("tontcap")]
      public ActionResult<Video> PostVideo(Video Video)
      {
          videoService.Create(Video);
      return CreatedAtAction("GetVideo", new { id = Video.Id }, Video);
  }

4、效果展示?

RabbitMQ宕机情况

步骤

1、将RabbitMQ直接关闭

事件消息无法发送,存储到内存缓存中

2、当将RabbitMQ启动后,消息正常发送

内部使用定时器轮询机制实现

AggregateService宕机情况

AggregateService 执行业务成功,发送消息前宕机

使用本地消息表解决(思想:持久化操作)

条件

1、本地消息表

步骤

1、在RuanMou.MicroService.AggregateService服务中

1.1 创建Context文件,然后在Context文件夹内创建AggregateContext

/// <summary>
    /// Aggregate服务上下文
    /// </summary>
    public class AggregateContext : DbContext
    {
        public AggregateContext(DbContextOptions<AggregateContext> options) : base(options)
        {
        }
}

1.2 在appsettings.json中添加

{
  "ConnectionStrings": {
    "DefaultConnection": "Data Source=.;Initial Catalog=aggregateservice;Persist Security Info=True;User ID=sa;Password=tony"
  }
}

1.3 在startup.cs中添加消息持久化

         // 9、注册上下文到IOC容器
            services.AddDbContext<AggregateContext>(options =>
            {
                options.UseSqlServer(Configuration.GetConnectionString("DefaultConnection"));
            });
        // 8、添加事件总线cap
        services.AddCap(x => {
            // 8.1 使用EntityFramework进行存储操作
            x.UseEntityFramework<AggregateContext>();
            // 8.2 使用sqlserver进行事务处理
            x.UseSqlServer(Configuration.GetConnectionString("DefaultConnection"));
        // 8.2 使用RabbitMQ进行事件中心处理
        x.UseRabbitMQ(rb => {
            rb.HostName = "localhost";
            rb.UserName = "guest";
            rb.Password = "guest";
            rb.Port = 5672;
            rb.VirtualHost = "/";
        });
    });

1.4测试演示效果

数据库中多了两张表

技术图片

当业务执行成功,发送消息时,聚合微服务宕机,消息被持久化到数据库

当重启聚合微服务时,消息发送成功,被成功消费

1.5 原理

1、定时器 消息重试

2、幂等性 一个函数每次都是相同的结果,状态只有一个

VideoService宕机情况

VideoService接受消息失败

当VideoService直接宕机的时候接受消息失败,

然后重启VideoService消息消费成功

VideoService接受消息成功执行失败

条件

1、本地消息表

步骤

1、在RuanMou.MicroService.Core项目中

1.1 安装SqlServer

Nuget DotNetCore.CAP.SqlServer

2、在RuanMou.MicroService.VideoService项目中

2.1 在startup.cs中添加消息持久化

    // 8、添加事件总线cap
    services.AddCap(x => {
        // 8.1 使用EntityFramework进行存储操作
        x.UseEntityFramework<AggregateContext>();
        // 8.2 使用sqlserver进行事务处理
        x.UseSqlServer(Configuration.GetConnectionString("DefaultConnection"));
    // 8.2 使用RabbitMQ进行事件中心处理
    x.UseRabbitMQ(rb => {
        rb.HostName = "localhost";
        rb.UserName = "guest";
        rb.Password = "guest";
        rb.Port = 5672;
        rb.VirtualHost = "/";
    });
});

2.2 效果展示

数据库多了两张表

技术图片

2.3 原理

1、定时器 消息重试

2、幂等性 一个函数每次都是相同的结果,状态只有一个

消息重试完还是消费失败情况

使用人工干预实现

条件

1、Dashboard -- 后台管理页面

步骤

1、在RuanMou.MicroService.Core项目中

1.1 安装Dashboard

Nuget DotNetCore.CAP.Dashboard

2、在RuanMou.MicroService.VideoService项目中

2.1 在startup.cs中添加Dashboard

// 8、添加事件总线cap
services.AddCap(x => {
    // 8.1 使用EntityFramework进行存储操作
    x.UseEntityFramework<AggregateContext>();
    // 8.2 使用sqlserver进行事务处理
    x.UseSqlServer(Configuration.GetConnectionString("DefaultConnection"));
// 8.3 使用RabbitMQ进行事件中心处理
x.UseRabbitMQ(rb => {
    rb.HostName = "localhost";
    rb.UserName = "guest";
    rb.Password = "guest";
    rb.Port = 5672;
    rb.VirtualHost = "/";
});

// 8.4添加cap后台监控页面
    x.UseDashboard();
});

2.2 运行打开cap后台监控页面

http://localhost:5007/cap

对于发送失败的消息进行重复发送

对于消费失败的消息进行重复消费

 

以上是关于事件总线--课件的主要内容,如果未能解决你的问题,请参考以下文章

计算机组成原理答疑--来源于南京大学课件

Android中使用事件总线的优缺点分别是啥?

事件总线框架-OTTO

Android事件总线

RxJava 作为事件总线被多次调用,即使只触发一次

android事件总线(eventbus)设计与实现