基于.net core微服务（Consul、Ocelot、Docker、App.Metrics+InfluxDB+Grafana、Exceptionless、数据一致性、Jenkins）

/v1/catalog/register : Registers a new node, service, or check/v1/catalog/deregister : Deregisters a node, service, or check/v1/catalog/datacenters : Lists known datacenters/v1/catalog/nodes : Lists nodes in a given DC/v1/catalog/services : Lists services in a given DC/v1/catalog/service/ : Lists the nodes in a given service/v1/catalog/node/ : Lists the services provided by a node

health endpoints：health endpoints用来查询健康状况相关信息，该功能从catalog中单独分离出来

/v1/healt/node/: 返回node所定义的检查，可用参数?dc=/v1/health/checks/: 返回和服务相关联的检查，可用参数?dc=/v1/health/service/: 返回给定datacenter中给定node中service/v1/health/state/: 返回给定datacenter中指定状态的服务，state可以是"any", "unknown", "passing", "warning", or "critical"，可用参数?dc=

session endpoints：session endpoints用来create、update、destory、query sessions

/v1/session/create: Creates a new session/v1/session/destroy/: Destroys a given session/v1/session/info/: Queries a given session/v1/session/node/: Lists sessions belonging to a node/v1/session/list: Lists all the active sessions

/v1/acl/create: Creates a new token with policy/v1/acl/update: Update the policy of a token/v1/acl/destroy/: Destroys a given token/v1/acl/info/: Queries the policy of a given token/v1/acl/clone/: Creates a new token by cloning an existing token/v1/acl/list: Lists all the active tokens

event endpoints：event endpoints用来fire新的events、查询已有的events

/v1/event/fire/: 触发一个新的event，用户event需要name和其他可选的参数，使用PUT方法/v1/event/list: 返回agent知道的events

status endpoints：status endpoints用来或者consul 集群的信息

/v1/status/leader : 返回当前集群的Raft leader/v1/status/peers : 返回当前集群中同事

2.6、使用consul

启动语法：

consul agent -server -datacenter=数据中心名称 -bootstrap -data-dir 数据存放路径 -config-file 配置文件路径 -ui-dir UI存放路径 -node=n1 -bind 本机IP

注册成Windows服务

sc.exe create "Consul" binPath= "E:\Consul\consule.exe agent -server -datacenter=数据中心名称 -bootstrap -data-dir 数据存放路径 -config-file 配置文件路径 -ui-dir UI存放路径 -node=n1 -bind 本机IP"

示例：

consul agent -server -datacenter=dc1 -bootstrap -data-dir /tmp/consul -config-file ./conf -ui-dir ./dist -node=n1 -bind 127.0.0.1

查看集群成员

consul members

把192.168.1.126加入集群

consul join 192.168.1.126

查看节点raft信息

consul operator raft list-peers

2.7、项目实例

│ consul.exe│ ├─conf│ service.json│ watchs.json│ xacl.json│ ├─data├─dist

service.json（服务发现配置）：

{ "encrypt": "7TnJPB4lKtjEcCWWjN6jSA==", //加密秘钥 "services": [{ "id": "BasicsService", //服务id "name": "BasicsService", //服务名称 "tags": ["BasicsService"], //服务标签 "address": "192.168.103.203", //服务地址 "port": 6801, //端口 "checks": [{ "id": "BasicsServiceCheck", //检查id "name": "BasicsServiceCheck", //检查名称 "" //检车接口地址 "interval": "10s", //检查周期 "tls_skip_verify": false, //跳过验证 "method": "GET", //检查请求方法 "timeout": "1s" //请求超时时间 }] }, { "id": "InvoicingService", //服务id "name": "InvoicingService", //服务名称 "tags": ["InvoicingService"], //服务标签 "address": "192.168.103.203", //服务地址 "port": 6802, //端口 "checks": [{ "id": "InvoicingServiceCheck", //检查id "name": "InvoicingServiceCheck", //检查名称 "" //检车接口地址 "interval": "10s", //检查周期 "tls_skip_verify": false, //跳过验证 "method": "GET", //检查请求方法 "timeout": "1s" //请求超时时间 }] } ]}

watchs.json（服务监控配置）：

{ "watches": [{ "type": "checks", //监控触发类型 "handler_type": " //异常通知类型 "state": "critical", //监控触发状态 "{ "path": " //通知地址 "method": "POST", //通知请求方式 "timeout": "10s", //通知超时时间 "header": { "Authorization": ["Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJodHRwOi8vc2NoZW1hcy54bWxzb2FwLm9yZy93cy8yMDA1LzA1L2lkZW50aXR5L2NsYWltcy9uYW1lIjoiZ3N3IiwiaHR0cDovL3NjaGVtYXMubWljcm9zb2Z0LmNvbS93cy8yMDA4LzA2L2lkZW50aXR5L2NsYWltcy9yb2xlIjoiYWRtaW4iLCJodHRwOi8vc2NoZW1hcy5taWNyb3NvZnQuY29tL3dzLzIwMDgvMDYvaWRlbnRpdHkvY2xhaW1zL2V4cGlyYXRpb24iOiIyMDIyLzEyLzMxIDEyOjM2OjEyIiwibmJmIjoxNTE0Njk0OTcyLCJleHAiOjE1MTQ3MzA5NzIsImlzcyI6ImdzdyIsImF1ZCI6ImdzdyJ9.jPu1yZ8jORN5QgCuPV50sYOKvX88GLSDiRX_0fpEzU4"] } //请求头 } }]}

分别启动 192.168.103.203 、 192.168.103.207 上的应用基础和进销存服务，然后再启动Consul，我们让 192.168.103.203 作为主Consul第一台service（192.168.103.203）：

consul agent -server -datacenter=dc1 -bootstrap -data-dir ./data -config-file ./conf -ui-dir ./dist -node=n1 -bind 192.168.103.203

第二台service（192.168.103.207）：

consul agent -server -datacenter=dc1 -data-dir ./data -config-file ./conf -ui-dir ./dist -node=n2 -bind 192.168.103.207

然后可以通过访问192.168.103.203:8500进入UI页面查看信息

client

consul agent -datacenter=dc1 -data-dir /tmp/consul -node cn1

Mac OX系统，进入consul所在目录执行：

Sudo scp consul /usr/local/bin/

2.8、Consul DNS

DnsAgent.exe作为DNS工具

[ { "Pattern": "^.*\\.consul$", "NameServer": "127.0.0.1:8600", "QueryTimeout": 1000, "CompressionMutation": false }]

访问地址：Ocelot是一堆特定顺序的中间件。

3.1、Ocelot使用

安装Ocelot

Install-Package Ocelot

public static IWebHost BuildWebHost(string[] args){ IWebHostBuilder builder = new WebHostBuilder(); //注入WebHostBuilder return builder.ConfigureServices(service => { service.AddSingleton(builder); }) //加载configuration配置文人年 .ConfigureAppConfiguration(conbuilder => { conbuilder.AddJsonFile("appsettings.json"); conbuilder.AddJsonFile("configuration.json"); }) .UseContentRoot(Directory.GetCurrentDirectory()) .UseKestrel() .UseUrls(" .UseStartup() .Build();}

public void ConfigureServices(IServiceCollection services){ //注入配置文件 services.AddOcelot(Configuration);}public void Configure(IApplicationBuilder app, IHostingEnvironment env){ //添加中间件 app.UseOcelot().Wait();}

{ "ReRoutes": [ //路由配置 { "DownstreamPathTemplate": "/{url}", //下游请求路由 "DownstreamScheme": " //下游请求方式，有 "DownstreamHostAndPorts": [ //下游请求的host和端口，为了配合负载均衡，可以配置多项 { "Host": "localhost", "Port": 6801 } ], "UpstreamPathTemplate": "/basics/{url}", //上游请求路由 "UpstreamHttpMethod": [ "Get", "Post", "Delete", "Put" ], //上游请求谓词 //"ServiceName": "BasicsService", //Consul中注册服务的名称 //"LoadBalancer": "RoundRobin", //负载均衡（可选）LeastConnection –请求空闲的Url RoundRobin – 轮询请求 NoLoadBalance – 无负载均衡 //"UseServiceDiscovery": true, //是否启用负载均衡 "ReRouteIsCaseSensitive": false, // "QoSOptions": { //熔断设置（可选） "ExceptionsAllowedBeforeBreaking": 3, //允许异常请求数 "DurationOfBreak": 10, //熔断时间，以秒为单位 "TimeoutValue": 5000 //请求超时数，以毫秒为单位 }, "HttpHandlerOptions": { // "AllowAutoRedirect": false, // "UseCookieContainer": false, // "UseTracing": false // }, "AuthenticationOptions": { // "AuthenticationProviderKey": "", // "AllowedScopes": [] // }, "RateLimitOptions": { //限流设置（可选） "ClientWhitelist": [ "admin" ], //白名单，不受限流控制的 "EnableRateLimiting": true, //是否启用限流 "Period": "1m", //统计时间段：1s, 2m, 3h, 4d "PeriodTimespan": 15, //间隔多少秒后可以重试 "Limit": 100 //设定时间段内允许的最大请求数 } }, { "DownstreamPathTemplate": "/{url}", "DownstreamScheme": " "DownstreamHostAndPorts": [ { "Host": "localhost", "Port": 6802 } ], "UpstreamPathTemplate": "/invoicing/{url}", "UpstreamHttpMethod": [ "Get", "Post", "Delete", "Put" ], //"ServiceName": "InvoicingService", //"LoadBalancer": "RoundRobin", //"UseServiceDiscovery": true, "ReRouteIsCaseSensitive": false, "QoSOptions": { "ExceptionsAllowedBeforeBreaking": 3, "DurationOfBreak": 10, "TimeoutValue": 5000 }, "HttpHandlerOptions": { "AllowAutoRedirect": false, "UseCookieContainer": false, "UseTracing": false }, "AuthenticationOptions": { "AuthenticationProviderKey": "", "AllowedScopes": [] }, "RateLimitOptions": { "ClientWhitelist": [ "admin" ], "EnableRateLimiting": true, "Period": "1m", "PeriodTimespan": 15, "Limit": 100 } }, { "DownstreamPathTemplate": "/{url}", "DownstreamScheme": " "DownstreamHostAndPorts": [ { "Host": "localhost", "Port": 6806 } ], "UpstreamPathTemplate": "/authentication/{url}", "UpstreamHttpMethod": [ "Get", "Post", "Delete", "Put" ], //"ServiceName": "AuthenticationService", //"LoadBalancer": "RoundRobin", //"UseServiceDiscovery": true, "ReRouteIsCaseSensitive": false, "QoSOptions": { "ExceptionsAllowedBeforeBreaking": 3, "DurationOfBreak": 10, "TimeoutValue": 5000 }, "HttpHandlerOptions": { "AllowAutoRedirect": false, "UseCookieContainer": false, "UseTracing": false }, "AuthenticationOptions": { "AuthenticationProviderKey": "", "AllowedScopes": [] }, "RateLimitOptions": { "ClientWhitelist": [ "admin" ], "EnableRateLimiting": true, "Period": "1m", "PeriodTimespan": 15, "Limit": 100 } } ], "GlobalConfiguration": { //全局设置 "ServiceDiscoveryProvider": {//Consul服务地址，用于上方的服务发现 "Host": "localhost", "Port": 8500 }, "RateLimitOptions": { //全局限流设置（可选） "ClientIdHeader": "clientid", //识别请求头，默认是 ClientId "QuotaExceededMessage": "access is denied", //被限流后，当请求过载时返回的提示消息 "HttpStatusCode": //600,当请求过载时返回的 "DisableRateLimitHeaders": false //此值指定是否禁用X-Rate-Limit和Rety-After标头 } }}

4、Docker

容器是一个打包了应用服务的环境。它是一个轻量级的虚拟机，每一个容器由一组特定的应用和必要的依赖库组成。

4.1、Docker-镜像常用命令

docker images:查看本地镜像，docker images ubu*，通配符查看docker inspect ubuntu:查看镜像详细信息docker search aspnetcore:搜索docker hub上符合要求的镜像docker pull microsoft/aspnetcore:拉取镜像，在run时不用从docker hub拉取docker rmi 镜像ID1 镜像ID2：删除镜像ID1和ID2，如果强制删除加-f

4.2、Docker-容器常用命令

docker create ubuntu:14.04：创建容器，处于停止状态docker ps：查看运行的容器，加-a查看所有容器。加-l查询出最后创建的容器，加-n=3查看最后创建的3个容器docker start 容器名：运行已存在的容器docker stop 容器名：停止容器docker rm 容器名：删除容器，docker rm $(docker ps -a -q)删除所有容器docker run -i -t --name ubuntu14 ubuntu:14.04 /bin/bash：运行一个ubuntu14.04的，带终端的容器，名字叫ubuntu14 ，-i用于打开容器的标准输入，-t让容器建立一个命令行终端docker run --name back_ubuntu14 -d ubuntu:14.04 /bin/sh -c "while true;do echo hello world;sleep 1;done"：-d是后台开容器docker attach 容器名：依附容器docker logs -f --tail=5 back_ubuntu14：查看最近的5条日志docker top 容器名：查看容器进程 docker inspect 容器名：查看容器信息，查看具体子项docker inspect --format='{{.NetworkSettings.IPAddress}}' back_ubuntu14docker export 容器名 >容器存储名称.tar：导出容器win powershell下 docker export 容器ID -o 名字.tardocker import 容器存储名称.tar：导入镜像docker commit -m="abc" --author="gsw" 容器ID 镜像名称：提交容器到本地镜像

4.4、Docker-Dockerfile

4.5、Docker生成asp- core镜像和运行

发布asp- core项目，并在发布文件夹下创建Dockerfile文件，复制下面内容

#父镜像FROM microsoft/aspnetcore#设置工作目录WORKDIR /app#复制发布文件到/app下COPY . /app#设置端口EXPOSE 80#使用dotnet XXXXXXXXX.dll来运行asp- core项目，注意大小写ENTRYPOINT ["dotnet", “XXXXXXXXX.dll"]

4.6、Docker生成asp- core镜像和运行

docker build -t xxxxxxxxxxx:latest .docker run -it -p 6801:6801 xxxxxxxxxxx:latest

注意：docker内部web的端口， 上述命令中，第二个端口为docker内web的端口。

5、App.Metrics+InfluxDB+Grafana

建议：建议在网关上进行监控，因为网关上监控可以监控所有

App.Metrics：​​influxdb​​​Grafana​​​database influxdbtest 创建数据库，同时 create user "user1" with password '123456'  创建用户配置Grafana，然后启动网关程序，登录localhost:3000查看监控信息，用户名密码是：admin

5.2、配置Grafana

5.2.1、配置数据源

5.2.2、配置Dashboard

我们采用模板导入模式，将项目引用 App.Metrics 并访问 App.Metrics 源地址：​​App.MetricsInstall-Package App.Metrics.AspNetCore.EndpointsInstall-Package App.Metrics.AspNetCore.ReportingInstall-Package App.Metrics.AspNetCore.TrackingInstall-Package App.Metrics.Reporting.InfluxDB

"InfluxDB": { "IsOpen": true,//是否开启监控 "DataBaseName": "influxdbtest",//数据库名称 "ConnectionString": " "username": "user1",//用户名 "password": "123456",//密码 "app": "HIS",//自定义名字 "env": "Ocelot"//自定义环境}

public void ConfigureServices(IServiceCollection services){ ... #region Metrics监控配置 string IsOpen = Configuration.GetSection("InfluxDB:IsOpen").Value.ToLower();//是否打开 if (IsOpen == "true") { string database = Configuration.GetSection("InfluxDB")["DataBaseName"];//数据库名称 string InfluxDBConStr = Configuration.GetSection("InfluxDB")["ConnectionString"];//数据库地址 string app = Configuration.GetSection("InfluxDB")["app"];//自定义名字 string env = Configuration.GetSection("InfluxDB")["env"];//自定义环境 string username = Configuration.GetSection("InfluxDB")["username"];//用户名 string password = Configuration.GetSection("InfluxDB")["password"];//密码 var uri = new Uri(InfluxDBConStr); //配置metrics var metrics = AppMetrics.CreateDefaultBuilder() .Configuration.Configure( options => { options.AddAppTag(app); options.AddEnvTag(env); }) .Report.ToInfluxDb( options => { options.InfluxDb.BaseUri = uri; options.InfluxDb.Database = database; options.InfluxDb.UserName = username; options.InfluxDb.Password = password; options.HttpPolicy.BackoffPeriod = TimeSpan.FromSeconds(30); options.HttpPolicy.FailuresBeforeBackoff = 5; options.HttpPolicy.Timeout = TimeSpan.FromSeconds(10); options.FlushInterval = TimeSpan.FromSeconds(5); }) .Build(); services.AddMetrics(metrics);//注入metrics services.AddMetricsReportScheduler();//报表 services.AddMetricsTrackingMiddleware();//追踪 services.AddMetricsEndpoints(); //终点 } #endregion ...}

public async void Configure(IApplicationBuilder app, IHostingEnvironment env){ #region 使用中间件Metrics string IsOpen = Configuration.GetSection("InfluxDB")["IsOpen"].ToLower(); if (IsOpen == "true") { app.UseMetricsAllMiddleware(); app.UseMetricsAllEndpoints(); } #endregion //使用中间件 //await app.UseOcelot();}

接下来就可以进行追踪了

5.4、APM-Grafana告警

[smtp]enabled = truehost = smtp.163.com:25user =ego_it# If the password contains # or ; you have to wrap it with triple quotes. Ex """#password;"""password =******cert_file =key_file =skip_verify = falsefrom_address = ego_it@163.comfrom_name = Grafanaehlo_identity =

6、Exceptionless

在线方式​​​web接口。Elasticsearch是用Java开发的，并作为Apache许可条款下的开放源码发布，是当前流行的企业级搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。Kibana是一个开源的分析与可视化平台，设计出来用于和Elasticsearch一起使用的。你可以用kibana搜索、查看、交互存放在Elasticsearch索引里的数据，使用各种不同的图表、表格、地图等kibana能够很轻易地展示高级数据分析与可视化。

6.1、创建组织

6.2、创建项目

6.3、集成Exceptionless 客户端

Install-Package Exceptionless.AspNetCore

通过 API 密钥执行  app.UseExceptionless("Qa3OzvEJC9FXo9SdwwFBv6bAkVbjWQKbV6hhtYEM")  方法

6.4、示例代码

#region Exceptionless测试try{ ExceptionlessClient.Default.SubmitLog("调试Exceptionless.Logging.LogLevel.Debu", Exceptionless.Logging.LogLevel.Debug); ExceptionlessClient.Default.SubmitLog("错误Exceptionless.Logging.LogLevel.Error", Exceptionless.Logging.LogLevel.Error); ExceptionlessClient.Default.SubmitLog("大错Exceptionless.Logging.LogLevel.fatal", Exceptionless.Logging.LogLevel.Fatal); ExceptionlessClient.Default.SubmitLog(" Exceptionless.Logging.LogLevel.Info", Exceptionless.Logging.LogLevel.Info); ExceptionlessClient.Default.SubmitLog(" Exceptionless.Logging.LogLevel.Off", Exceptionless.Logging.LogLevel.Off); ExceptionlessClient.Default.SubmitLog(" Exceptionless.Logging.LogLevel.Other", Exceptionless.Logging.LogLevel.Other); ExceptionlessClient.Default.SubmitLog(" Exceptionless.Logging.LogLevel.Trace", Exceptionless.Logging.LogLevel.Trace); ExceptionlessClient.Default.SubmitLog("Exceptionless.Logging.LogLevel.Warn", Exceptionless.Logging.LogLevel.Warn); var data = new Exceptionless.Models.DataDictionary(); data.Add("data1key", "data1value"); ExceptionlessClient.Default.SubmitEvent(new Exceptionless.Models.Event { Count = 1, Date = DateTime.Now, Data = data, Geo = "geo", Message = "message", ReferenceId = "referencelId", Source = "source", Tags = new Exceptionless.Models.TagSet() { "tags" }, Type = "type", Value = 1.2m }); ExceptionlessClient.Default.SubmitFeatureUsage("feature"); ExceptionlessClient.Default.SubmitNotFound("404 not found"); ExceptionlessClient.Default.SubmitException(new Exception("自定义异常")); throw new DivideByZeroException("throw DivideByZeroException的异常：" + DateTime.Now);}catch (Exception exc){ exc.ToExceptionless().Submit();}#endregion

6.5、本地部署

​​本地部署官方wiki​​

-​​Windows版本安装包​​，并进行解压，然后双击运行Start.bat即可

需要环境：

.NET 4.6Java 1.8+ (The JAVA_HOME environment variable must also be configured when using Windows.)IIS Express 8+PowerShell 3+

6.6、项目集成

注意：本地化不能再使用 app.UseExceptionless(apiKey: "tJxBWkCbgDLCMoKKqWII3Eyw4aJOsyOCgX26Yurm"); 形式来上传日志数据，应采用另外的方式：配置文件方式

"Exceptionless": { "ApiKey": "tJxBWkCbgDLCMoKKqWII3Eyw4aJOsyOCgX26Yurm", "ServerUrl": " "DefaultData": { }, "DefaultTags": [ "xplat" ], "Settings": { "FeatureXYZEnabled": false }}

然后修改 Startup.cs

public void Configure(IApplicationBuilder app, IHostingEnvironment env){ ... //app.UseExceptionless(apiKey: "tJxBWkCbgDLCMoKKqWII3Eyw4aJOsyOCgX26Yurm"); //上方的方法本地化不适用 app.UseExceptionless(Configuration); ...}

搞定

6.7、查询语法

示例

6.8、常见问题

Invoke-WebRequest : 请求被中止: 未能创建 SSL/TLS 安全通道。

elasticsearch-XXX”，因为该路径不存在。

解决方案：编辑Start-ElasticSearch.ps1，将所需的文件全部-下来，然后解压进行拷贝，如下图，然后在双击运行Start.bat即可

帮助类：

View Code

7、数据一致性

Ｃ：数据一致性(consistency)：如果系统对一个写操作返回成功，那么之后的读请求都必须读到这个新数据；如果返回失败，那么所有读操作都不能读到这个数据，对调用者而言数据具有强一致性(strong consistency) (又叫原子性 atomic、线性一致性 linearizable consistency)A：服务可用性(availability)：所有读写请求在一定时间内得到响应，可终止、不会一直等待P：分区容错性(partition-tolerance)：在网络分区的情况下，被分隔的节点仍能正常对外服务

7.1、最终一致性

可用性，可靠性，最终一致性：在微服务之间使用事件驱动通信和发布订阅系统实现最终一致性

强一致性：当更新操作完成之后，任何多个后续进程或者线程的访问都会返回最新的更新过的值。这种是对用户最友好的，就是用户上一次写什么，下一次就保证能读到什么。根据 CAP 理论，这种实现需要牺牲可用性。=> 在传统单体式应用中，大部分都是强一致性的应用，想想我们写过多少工作单元模式的Code？弱一致性：系统并不保证续进程或者线程的访问都会返回最新的更新过的值。系统在数据写入成功之后，不承诺立即可以读到最新写入的值，也不会具体的承诺多久之后可以读到。最终一致性：弱一致性的特定形式。系统保证在没有后续更新的前提下，系统最终返回上一次更新操作的值。在没有故障发生的前提下，不一致窗口的时间主要受通信延迟，系统负载和复制副本的个数影响。为保证可用性，互联网分布式架构中经常将强一致性需求转换成最终一致性的需求，并通过系统执行幂等性的保证，保证数据的最终一致性。

在微服务架构中，各个微服务之间通常会使用事件驱动通信和发布订阅系统实现最终一致性。

7.2、最终一致性-补偿机制

Polly：实现重试，熔断机制或提供后台任务调度实现补偿

7.3、幂等和防重

其任意多次执行对资源本身所产生的影响均与一次执行的影响相同。对重复删除或返回成功结果；防重可以在数据库级别处理也以以在MQ级别

7.4、MassTransit

MassTransit 是一个自由、开源、轻量级的消息总线, 用于使用. NET 框架创建分布式应用程序。MassTransit 在现有消息传输上提供了一组广泛的功能, 从而使开发人员能够友好地使用基于消息的会话模式异步连接服务。基于消息的通信是实现面向服务的体系结构的可靠和可扩展的方式。

官网地址：​​Service Bus。类似的国内开源组件则有园友savorboard（杨晓东）的​​CAP​​

7.5、最简单的发送/接收实例

这里以MassTransit + RabbitMQ为例子，首先请确保安装了RabbitMQ，如果没有安装，可以阅读我的​​RabbitMQ在Windows环境下的安装与使用​​去把RabbitMQ先安装到你的电脑上。另外，RabbitMQ的背景知识也有一堆，有机会也还是要了解下Exchange，Channel、Queue等内容。

准备两个控制台程序，一个为Sender（发送者），一个为Receiver（接收者），并分别通过NuGet安装MassTransit以及MassTransit.RabbitMQ

Install-Package MassTransitInstall-Package MassTransit.RabbitMQ

7.6、一对一的发布/订阅实例（类似于RabbitMQ的工作模式）

除了简单的发送/接收模式外，我们用的更多的是发布/订阅这种模式。

注意：发布方如果发布时没有订阅方，发布的数据将会丢失

public interface IEntity{ int ID { get; set; }}public class Entity:IEntity{ public int ID { get; set; } public string Name { get; set; } public DateTime Time { get; set; }}public class MyEntity:Entity{ public int Age { get; set; }}public class YouEntity{ public int ID { get; set; } public string Name { get; set; } public DateTime Time { get; set; } public int Age { get; set; }}

static void Main(string[] args){ Console.Title = "发布方"; var bus = Bus.Factory.CreateUsingRabbitMq(cfg => { var host = cfg.Host(new Uri("rabbitmq://localhost"), hst => { hst.Username("guest"); hst.Password("guest"); }); }); bus.Start(); do { Console.WriteLine("请出请按q,否则请按其他键！"); string value = Console.ReadLine(); if (value.ToLower() == "q") { break; } bus.Publish(new Entity() { ID = 1, Name = "张三", Time = DateTime.Now }); } while (true); bus.Stop();}

这里向RabbitMQ发布了两个不同类型的消息（Entity和IEntity）

class Program{ static void Main(string[] args) { Console.Title="订阅方"; var bus = Bus.Factory.CreateUsingRabbitMq(cfg => { var host = cfg.Host(new Uri("rabbitmq://localhost/"), hst => { hst.Username("guest"); hst.Password("guest"); }); cfg.ReceiveEndpoint(host, "mewyt", e => { e.Consumer(); e.Consumer(); e.Consumer(); }); cfg.ReceiveEndpoint(host, "mewyt1", e => { e.Consumer(); }); }); bus.Start(); Console.ReadLine(); bus.Stop(); }}public class IEntityConsumer : IConsumer{ public async Task Consume(ConsumeContext context) { await Console.Out.WriteLineAsync($"IEntityConsumer 类型 {context.Message.GetType()} {context.Message.ID}"); }}public class EntityConsumer : IConsumer{ public async Task Consume(ConsumeContext context) { await Console.Out.WriteLineAsync($"EntityConsumer 类型 {context.Message.GetType()} {context.Message.ID} {context.Message.Name} {context.Message.Time}"); }}public class MyEntityConsumer : IConsumer{ public async Task Consume(ConsumeContext context) { await Console.Out.WriteLineAsync($"MyEntityConsumer 类型 {context.Message.GetType()} {context.Message.ID} {context.Message.Name} {context.Message.Time} {context.Message.Age}"); }}public class YouEntityConsumer : IConsumer{ public async Task Consume(ConsumeContext context) { await Console.Out.WriteLineAsync($"YouEntityConsumer 类型 {context.Message.GetType()} {context.Message.ID} {context.Message.Name} {context.Message.Time} {context.Message.Age}"); }}

7.6、一对多的发布/订阅实例（队列名不同即可）

public class EntityA{ public string Name { get; set; } public DateTime Time { get; set; }}public class EntityB{ public string Name { get; set; } public DateTime Time { get; set; } public int Age { get; set; }}

class Program{ static void Main(string[] args) { var bus= Bus.Factory.CreateUsingRabbitMq(cfg => { var host = cfg.Host(new Uri("rabbitmq://localhost/"), hst => { hst.Username("guest"); hst.Password("guest"); }); }); do { Console.WriteLine("请出请按q,否则请按其他键！"); string value = Console.ReadLine(); if (value.ToLower() == "q") { break; } bus.Publish(new PSDemo_Entity.EntityA() { Name="张三", Time = DateTime.Now }); bus.Publish(new PSDemo_Entity.EntityB() { Name = "李四", Time = DateTime.Now,Age=22 }); } while (true); bus.Stop(); }}

class Program{ static void Main(string[] args) { Console.Title="订阅者A"; var bus= Bus.Factory.CreateUsingRabbitMq(cfg => { var host = cfg.Host(new Uri("rabbitmq://localhost/"), hst => { hst.Username("guest"); hst.Password("guest"); }); cfg.ReceiveEndpoint(host, "wytPSA", e => { e.Consumer(); e.Consumer(); }); }); bus.Start(); Console.ReadLine(); bus.Stop(); }}public class ConsumerA : IConsumer{ public async Task Consume(ConsumeContext context) { await Console.Out.WriteLineAsync($"订阅者A ConsumerA收到信息: {context.Message.Name} {context.Message.Time} 类型：{context.Message.GetType()}"); }}public class ConsumerB : IConsumer{ public async Task Consume(ConsumeContext context) { await Console.Out.WriteLineAsync($"订阅者A ConsumerB收到信息: {context.Message.Name} {context.Message.Time} 类型：{context.Message.GetType()}"); }}

class Program{ static void Main(string[] args) { Console.Title="订阅者B"; var bus = Bus.Factory.CreateUsingRabbitMq(cfg => { var host = cfg.Host(new Uri("rabbitmq://localhost/"), hst => { hst.Username("guest"); hst.Password("guest"); }); cfg.ReceiveEndpoint(host, "wytPSB", e => { e.Consumer(); }); }); bus.Start(); Console.ReadLine(); bus.Stop(); }}public class ConsumerA : IConsumer{ public async Task Consume(ConsumeContext context) { await Console.Out.WriteLineAsync($"订阅者B ConsumerA收到信息: {context.Message.Name} {context.Message.Time} 类型：{context.Message.GetType()}"); }}

7.7、带返回状态消息的示例

之前的例子都是发布之后，不管订阅者有没有收到以及收到后有没有处理成功（即有没有返回消息，类似于HTTP请求和响应），在MassTransit中提供了这样的一种模式，并且还可以结合GreenPipes的一些扩展方法实现重试、限流以及熔断机制。这一部分详见官方文档：​​interface IRequestEntity{ int ID { get; set; } string Name { get; set; }}public class RequestEntity : IRequestEntity{ public int ID { get; set; } public string Name { get; set; }}public interface IResponseEntity{ int ID { get; set; } string Name { get; set; } int RequestID { get; set; }}public class ResponseEntity : IResponseEntity{ public int ID { get; set; } public string Name { get; set; } public int RequestID { get; set; }}

class Program{ static void Main(string[] args) { Console.Title = "应答方"; var bus = Bus.Factory.CreateUsingRabbitMq(cfg => { var host = cfg.Host(new Uri("rabbitmq://localhost/"), hst => { hst.Username("guest"); hst.Password("guest"); }); cfg.ReceiveEndpoint(host, "request_response_wyt", e => { e.Consumer(); }); }); bus.Start(); Console.ReadLine(); bus.Stop(); }}public class RequestConsumer : IConsumer{ public async Task Consume(ConsumeContext context) { Console.ForegroundColor = ConsoleColor.Red; await Console.Out.WriteLineAsync($"收到请求id={context.Message.ID} name={context.Message.Name}"); Console.ResetColor(); var response = new ResponseEntity { ID = 22, Name = $"李四", RequestID = context.Message.ID }; Console.ForegroundColor = ConsoleColor.Green; Console.WriteLine($"应答ID={response.ID},Name={response.Name},RequestID={response.RequestID}"); Console.ResetColor(); context.Respond(response); }}

static void Main(string[] args){ Console.Title = "请求方"; var bus = Bus.Factory.CreateUsingRabbitMq(cfg => { var host = cfg.Host(new Uri("rabbitmq://localhost/"), hst => { hst.Username("guest"); hst.Password("guest"); }); //重试 cfg.UseRetry(ret => { ret.Interval(3, 10);// 消费失败后重试3次，每次间隔10s }); //限流 cfg.UseRateLimit(1000, TimeSpan.FromSeconds(100));// 1分钟以内最多1000次调用访问 //熔断 cfg.UseCircuitBreaker(cb => { cb.TrackingPeriod = TimeSpan.FromSeconds(60);//1分钟 cb.TripThreshold = 15;// 当失败的比例至少达到15%才会启动熔断 cb.ActiveThreshold = 10;// 当失败次数至少达到10次才会启动熔断 cb.ResetInterval = TimeSpan.FromMinutes(5);// 当在1分钟内消费失败率达到15%或调用了10次还是失败时，暂停5分钟的服务访问 }); }); bus.Start(); var serviceAddress = new Uri($"rabbitmq://localhost/request_response_wyt"); var client = bus.CreateRequestClient(serviceAddress, TimeSpan.FromHours(10)); // 创建请求客户端，10H之内木有回馈则认为是超时(Timeout) while (true) { Console.WriteLine("请出请按q,否则请按其他键！"); string value = Console.ReadLine(); if (value.ToLower() == "q") { break; } Task.Run(async () => { var request = new RequestEntity() { ID = 1, Name = "张三" }; var response = await client.Request(request); Console.WriteLine($"请求ID={request.ID},Name={request.Name}"); Console.WriteLine($"应签ID={response.ID},Name={response.Name},RequestID={response.RequestID}"); }).Wait(); }}

7.8、带Observer模式的发布/订阅示例

在某些场景中，我们需要针对一个消息进行类似于AoP（面向切面编程）或者监控的操作，比如在发送消息之前和结束后记日志等操作，我们可以借助MassTransit中的Observer模式来实现。（在MassTransit的消息接收中，可以通过两种模式来实现：一种是基于实现IConsumer接口，另一种就是基于实现IObserver接口）关于这一部分，详见官方文档：​​Program{ static void Main(string[] args) { Console.Title = "订阅方"; var bus = Bus.Factory.CreateUsingRabbitMq(cfg => { var host = cfg.Host(new Uri("rabbitmq://localhost/"), hst => { hst.Username("guest"); hst.Password("guest"); }); cfg.ReceiveEndpoint(host, "ObserverTest", e => { e.Consumer(); }); }); var observer = new ReceiveObserver(); var handle = bus.ConnectReceiveObserver(observer); bus.Start(); Console.ReadLine(); bus.Stop(); }}public class ReceiveObserver : IReceiveObserver{ public Task PreReceive(ReceiveContext context) { Console.WriteLine("------------------PreReceive--------------------"); Console.WriteLine(Encoding.Default.GetString(context.GetBody())); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; } public Task PostReceive(ReceiveContext context) { Console.WriteLine("-----------------PostReceive---------------------"); Console.WriteLine(Encoding.Default.GetString(context.GetBody())); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; } public Task PostConsume(ConsumeContext context, TimeSpan duration, string consumerType) where T : class { Console.WriteLine("------------------PostConsume--------------------"); Console.WriteLine($"ID={ (context.Message as Entity).ID},Name={(context.Message as Entity).Name},Time={(context.Message as Entity).Time}"); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; } public Task ConsumeFault(ConsumeContext context, TimeSpan elapsed, string consumerType, Exception exception) where T : class { Console.WriteLine("-----------------ConsumeFault---------------------"); Console.WriteLine($"ID={ (context.Message as Entity).ID},Name={(context.Message as Entity).Name},Time={(context.Message as Entity).Time}"); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; } public Task ReceiveFault(ReceiveContext context, Exception exception) { Console.WriteLine("----------------ReceiveFault----------------------"); Console.WriteLine(Encoding.Default.GetString(context.GetBody())); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; }}public class EntityConsumer : IConsumer{ public async Task Consume(ConsumeContext context) { await Console.Out.WriteLineAsync($"IEntityConsumer 类型 {context.Message.GetType()} {context.Message.ID} {context.Message.Age} {context.Message.Name} {context.Message.Time}"); }}public class Entity{ public int ID { get; set; } public int Age { get; set; } public string Name { get; set; } public DateTime Time { get; set; }}

class Program{ static void Main(string[] args) { Console.Title = "发布方"; var bus = Bus.Factory.CreateUsingRabbitMq(cfg => { var host = cfg.Host(new Uri("rabbitmq://localhost"), hst => { hst.Username("guest"); hst.Password("guest"); }); }); var observer = new SendObserver(); var handle = bus.ConnectSendObserver(observer); var observer1 = new PublishObserver(); var handle1 = bus.ConnectPublishObserver(observer1); bus.Start(); do { Console.WriteLine("请出请按q,否则请按其他键！"); string value = Console.ReadLine(); if (value.ToLower() == "q") { break; } bus.Publish(new Entity() { ID = 1, Age = 10, Name = "张三", Time = DateTime.Now }); } while (true); bus.Stop(); }}public class PublishObserver : IPublishObserver{ public Task PrePublish(PublishContext context) where T : class { Console.WriteLine("------------------PrePublish--------------------"); Console.WriteLine($"ID={ (context.Message as Entity).ID},Name={(context.Message as Entity).Name},Time={(context.Message as Entity).Time}"); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; } public Task PostPublish(PublishContext context) where T : class { Console.WriteLine("------------------PostPublish--------------------"); Console.WriteLine($"ID={ (context.Message as Entity).ID},Name={(context.Message as Entity).Name},Time={(context.Message as Entity).Time}"); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; } public Task PublishFault(PublishContext context, Exception exception) where T : class { Console.WriteLine("------------------PublishFault--------------------"); Console.WriteLine($"ID={ (context.Message as Entity).ID},Name={(context.Message as Entity).Name},Time={(context.Message as Entity).Time}"); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; }}public class SendObserver : ISendObserver{ public Task PreSend(SendContext context) where T : class { Console.WriteLine("------------------PreSend--------------------"); Console.WriteLine($"ID={ (context.Message as Entity).ID},Name={(context.Message as Entity).Name},Time={(context.Message as Entity).Time}"); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; } public Task PostSend(SendContext context) where T : class { Console.WriteLine("------------------PostSend--------------------"); Console.WriteLine($"ID={ (context.Message as Entity).ID},Name={(context.Message as Entity).Name},Time={(context.Message as Entity).Time}"); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; } public Task SendFault(SendContext context, Exception exception) where T : class { Console.WriteLine("------------------SendFault--------------------"); Console.WriteLine($"ID={ (context.Message as Entity).ID},Name={(context.Message as Entity).Name},Time={(context.Message as Entity).Time}"); Console.WriteLine("--------------------------------------"); return Task.CompletedTask; }}

7.9、数据一致性示例

详见：​​是一款流行的开源持续集成（CI）与持续部署（CD）工具，广泛用于项目开发，具有自动化构建、测试和部署等功能。

使用Jenkins的目的在于：

（1）持续、自动地构建/测试软件项目。 　　（2）监控软件开放流程，快速问题定位及处理，提升开发效率。

8.1、Jenkins-与安装

这里我们-Windows版本的

安装完成后会提示我们解锁 Jenkins

这里选择安装推荐的插件

创建管理账户 => 也可以直接使用admin账户继续

配置Jenkins端口，默认8080，最好不要使用8080端口

修改Jenkins服务端口，改为8080-->8081

修改安装目录下 jenkins.xml 文件

然后重启Jenkins服务

8.2、持续集成Asp.Net Core项目

cd CITestcd CITestdotnet restoredotnet builddotnet publish -o "bin\Debug\netcoreapp2.0\publish"cd bin\Debug\netcoreapp2.0\publishdocker rm clitest -fdocker rmi clitest -fdocker build -t clitest:latest .docker run -p 4555:4555 -d --name clitest clitest:latest

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。