读写分离

基于Nebula的项目启用数据库 读写分离 需要2个步骤：

• 在配置服务中注册只读库
• 在调用代码中指定参数使用读库

在配置服务中注册只读库

这个过程其实就是创建一个与【普通连接】类似的【只读连接】配置， 以下图为例：

示例解释：

• yunwei-s1 是一个可读可写的数据库连接配置
• yunwei-s1_readonly 是一个只读的数据库连接配置
• _readonly 这个后缀是固定的（一个约定），用于配置服务查找只读连接

在调用代码中指定参数使用读库

以下是一个 Action 的示例代码：

public async Task<string> Test2()
{
    // 连接到  与当前用户对应的 【只读/租户库】
    using( DbContext dbContext = this.CreateTenantConnection(tenantId:"xxxxx", readonlyDB:true) ) {

        return await dbContext.CPQuery.Create("select now()").ExecuteScalarAsync<string>();
    }
}

在这个示例中，调用 CreateTenantConnection 方法时第2个参数 readonlyDB:true 就表示使用【只读库】

也可以直接调用 DbConnManager.CreateTenant 方法

/// <summary>
/// 根据 租户ID 创建对应的SQL数据库连接
/// </summary>
/// <param name="tenantId">租户ID</param>
/// <param name="readonlyDB">是否连接【只读库】</param>
/// <returns></returns>
public static DbContext CreateTenant(string tenantId, bool readonlyDB = false)

InfluxClient也支持类似的只读库连接：

/// <summary>
/// 构造InfluxClient实例，并根据 租户ID 创建对应的连接
/// </summary>
/// <param name="tenantId">租户ID</param>
/// <param name="readonlyDB">是否连接【只读库】</param>
/// <returns></returns>
public static InfluxClient CreateTenant(string tenantId, bool readonlyDB = false)

数据库分库

数据库分库（或者叫“拆分数据库”）通常有2种做法：

• 按业务边界垂直拆分
• 按用户水平拆分

Nebula同时支持这2种做法：

• 垂直拆分：在Nebula中称为“应用库”拆分，也就是某个应用单独操作某个数据库
• 水平拆分：Nebula支持按客户来拆分数据库（客户隔离），拆分后的数据库被称为“租户库”

注册应用库

应用库在配置服务中就是一个普通的数据库连接，如下图所示：

在表格中，Name 列的名称将在代码中被引用。

连接到应用库

public async Task<string> Test4()
{
    // 连接到 某个特定的 【应用库】
    // xxx_Db 是一个连接名称，在【配置服务】中有对应的连接设置

    using( DbContext dbContext = this.CreateAppDbConnection("xxx_Db") ) {

        return await dbContext.CPQuery.Create("select now()").ExecuteScalarAsync<string>();
    }
}

有2种方法连接到【应用库】

• 调用 BaseController.CreateAppDbConnection("name") 方法
• 调用 DbConnManager.CreateAppDb("name") 方法

注册租户库

这里又分为2个步骤：

• 为租户的数据库实例注册数据库连接
• 为每个租户指定连接名字

这里假设所有租户的数据库部署在3个数据库实例上，
我们就可以分别为它们创建3个数据库连接：yunwei-s1, yunwei-s2, yunwei-s3

在tenantconn表中，可以为每个租户指定与数据库实例的映射关系。

连接到租户库

可参考下面示例代码：

public async Task<string> Test1()
{
    // 从请求中获取到的租户信息
    // 通常是登录操作，此时用户还没有经过登录验证，这时候需要在提交登录数据时把 租户ID 一起发送到服务端
    // 服务端根据 租户ID 就可以连接到相应的租户库
    string tenantId = "xxxxxxxxxx";

    using( DbContext dbContext = this.CreateTenantConnection(tenantId) ) {  // 连接到 指定的【租户库】

        return await dbContext.CPQuery.Create("select now()").ExecuteScalarAsync<string>();
    }
}

public async Task<string> Test2()
{
    // 连接到  与当前用户对应的 【租户库】
    using( DbContext dbContext = this.CreateTenantConnection() ) {

        return await dbContext.CPQuery.Create("select now()").ExecuteScalarAsync<string>();
    }
}

或者直接调用 DbConnManager.CreateTenant 方法：

/// <summary>
/// 根据 租户ID 创建对应的SQL数据库连接
/// </summary>
/// <param name="tenantId">租户ID</param>
/// <param name="readonlyDB">是否连接【只读库】</param>
/// <returns></returns>
public static DbContext CreateTenant(string tenantId, bool readonlyDB = false)

使用达梦数据库

ClownFish已支持达梦数据库，具体使用方法可参考本文档。

1，引用 NuGet 包

在csproj项目文件中添加

<PackageReference Include="ClownFish.DmProvider" Version="8.3.1.32690" />
<PackageReference Include="ClownFish.net" Version="9.25.1105.1" />

2，配置数据库连接

在ClownFish.App.config中添加：

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
    <appSettings>
    </appSettings>

    <dbConfigs>
        <add name="dm1" dbType="DaMeng" server="DamengHost" port="5237" database="MyNorthwind" uid="SYSDBA" pwd="xxxxxxx" />
    </dbConfigs>
</configuration>

说明：也可以在配置服务中注册连接。

3，CRUD开发

这个过程与使用 SQLSERVER/MySQL/PostgreSQL 没有什么差别。
以下是实际的测试代码片段：

using DbContext db = DbContext.Create("dm1");

var args = new { CategoryID = 1 };

List<Product> list = db.CPQuery.Create("select * from Products where CategoryID = @CategoryID", args).ToList<Product>();

Console.WriteLine(list.ToXml());

using DbContext db = DbContext.Create("dm1");

db.CPQuery.Create("truncate table TestGuid").ExecuteNonQuery();

string insertSQL = "insert into TestGuid(RowGuid, IntValue) values(@RowGuid, @IntValue)";
for( int i = 0; i < 1000; i++ ) {

    var args = new {
        RowGuid = db.NewSeqGuid(),
        IntValue = i + 1
    };

    db.CPQuery.Create(insertSQL, args).ExecuteNonQuery();
}

using DbContext db = DbContext.Create("dm1");

Product product = new();
product.LoadDefaultValues();

// insert
int newId = (int)product.Insert(db, true);

// query
Product product2 = db.GetByKey<Product>(newId);
Assert.IsNotNull(product2);


Product product3 = new Product{ ... };
// update
int effect = db.Update(product3);
Assert.AreEqual(1, effect);


Product product4 = db.GetByKey<Product>(newId);
Assert.IsNotNull(product4);


// delete
db.Delete<Product>(newId);

Product product5 = db.GetByKey<Product>(newId);
Assert.IsNull(product5);  

初始化数据访问层

基于Nebula开发的项目，可以不用关注本文所介绍的内容。

本文仅供直接使用 ClownFish 的项目参考。

引用数据库驱动包

首先需要在项目文件中引用所需的驱动包，例如下面支持了5种数据库

<ItemGroup>
    <PackageReference Include="System.Data.SqlClient" Version="4.8.5" />
    <PackageReference Include="MySqlConnector" Version="2.2.5" />
    <PackageReference Include="Npgsql" Version="6.0.4" />
    <PackageReference Include="System.Data.SQLite.Core" Version="1.0.114.3" />
    <PackageReference Include="ClownFish.DmProvider" Version="8.3.1.32690" />
</ItemGroup>

说明：

• 需要使用哪种数据库就引用哪个，ClownFish会根据DLL文件自动加载

ClownFish 的初始化

请确保以下代码能在程序启动时运行

ClownFishInit.InitDAL();

对 SQLSERVER 的特别支持

• .net framework 项目
  • 内置支持，固定使用 System.Data.SqlClient
  • 不再需要引用任何SqlClient相关包，也不需要额外的初始化代码
• .net6/7/8 项目
  • 支持2种驱动包：System.Data.SqlClient, Microsoft.Data.SqlClient
  • 引用哪个就使用哪个

对 MySQL 的特别支持

ClownFish 支持2种 MySQL 驱动类库，具体内容可参考：MySQL异步

例如：

<PackageReference Include="MySql.Data" Version="8.0.29" />
<PackageReference Include="MySqlConnector" Version="2.2.5" />

引用哪个就使用哪个！

使用EntityFrameworkCore

虽然Nebula默认使用ClownFish中提供的数据访问模块，但也支持使用其它的数据访问模块。

本文将介绍如何在Nebula中使用EntityFrameworkCore

添加NuGet包

例如：

<PackageReference Include="Microsoft.EntityFrameworkCore" Version="7.0.2" />
<PackageReference Include="Pomelo.EntityFrameworkCore.MySql" Version="7.0.0" />
<PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="7.0.0" />
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="7.0.0" />

定义EF-DbContext继承类

示例代码如下

public class MyNorthwindEfDbContext : Microsoft.EntityFrameworkCore.DbContext
{
    public DbSet<Product> Product { get; set; }

  
    private readonly DbConfig _dbConfig;

    public MyNorthwindEfDbContext(DbConfig dbConfig)
    {
        _dbConfig = dbConfig;
    }

    protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
    {
        string connectionString = _dbConfig.GetConnectionString(true);

        if( _dbConfig.DbType == DatabaseType.SQLSERVER )
            optionsBuilder.UseSqlServer(connectionString);

        else if( _dbConfig.DbType == DatabaseType.MySQL ) {

            ServerVersion version = ServerVersion.AutoDetect(connectionString);
            optionsBuilder.UseMySql(connectionString, version, null);
        }

        else if( _dbConfig.DbType == DatabaseType.PostgreSQL )
            optionsBuilder.UseNpgsql(connectionString);

        else if( _dbConfig.DbType == DatabaseType.DaMeng )
            optionsBuilder.UseDm(connectionString);

        else
            throw new NotSupportedException();
    }
}

在代码中使用

以下代码来自 Controller 中的示例

[Route("search/{index}/{size}.aspx")]
public PageListResult<Product> Search([FromBody] ProductSearchParam product, int index, int size)
{
    PageListResult<Product> result = new PageListResult<Product>();
    result.PagingInfo = new PagingInfo();
    result.PagingInfo.PageIndex = index;
    result.PagingInfo.PageSize = size;

    DbConfig dbConfig = DbConnManager.GetAppDbConfig("conn-name");

    using( MyNorthwindEfDbContext dbContext = new MyNorthwindEfDbContext(dbConfig) ) {

        var query = dbContext.Product.Select(x => x);

        if( product.CategoryID > 0 )
            query = query.Where(x => x.CategoryID == product.CategoryID);

        if( product.UnitPrice > 0 )
            query = query.Where(x => x.UnitPrice > product.UnitPrice);

        if( product.Unit.HasValue() )
            query = query.Where(x => x.Unit == product.Unit);

        if( product.ProductName.HasValue() )
            query = query.Where(x => EF.Functions.Like(x.ProductName, product.ProductName + "%"));

        query = query.OrderBy(x => x.ProductID);

        result.Data = query.ToList();
        return result;
    }
}

性能日志展示如下：

扩展-数据库种类

ClownFish只支持几种常用数据库，如果没有你需要的数据库，请自行扩展。

本文演示对 人大金仓-KingbaseES 数据库的支持

引用数据库驱动包

首先我们引用官方的客户端驱动

<ItemGroup>
    <PackageReference Include="Kdbndp_V9" Version="8.0.1.906" />
</ItemGroup>

实现 BaseClientProvider 抽象类

完整代码如下：

public class KingbaseESClientProvider : BaseClientProvider
{
    public static readonly BaseClientProvider Instance = new KingbaseESClientProvider();

    #region 定义2个"常量"，可以避免在其它代码中出现“硬编码”。
    public static readonly string ProviderName = "Kdbndp";

    // 由于 DatabaseType 是枚举，无法扩展，所以只能使用“强转”方式
    public static readonly DatabaseType DatabaseTypeKingbaseES = (DatabaseType)7777;
    #endregion


    public static void RegisterProvider()
    {
        // Npgsql 6.0 对时间戳的映射方式进行了一些重要更改
        // https://www.npgsql.org/doc/types/datetime.html#timestamps-and-timezones
        AppContext.SetSwitch("Kdbndp.EnableLegacyTimestampBehavior", true);
        AppContext.SetSwitch("Kdbndp.DisableDateTimeInfinityConversions", true);

        DbClientFactory.RegisterProvider(ProviderName, Instance);
    }

    public override DatabaseType DatabaseType => DatabaseTypeKingbaseES;

    public override DbProviderFactory ProviderFactory => Kdbndp.KdbndpFactory.Instance;

    public override string GetObjectFullName(string symbol)
    {
        // https://help.kingbase.com.cn/v8/faq/faq-new/sql.html#id12
        return "\"" + symbol + "\"";
    }


    public override CPQuery GetNewIdQuery(CPQuery query, object entity)
    {
        return query + "; SELECT lastval();";  // 参考 PostgreSql 的实现
    }


    public override bool IsDuplicateInsertException(Exception ex)
    {
        if( ex is Kdbndp.KingbaseException ex2 ) {
            return ex2.SqlState == "23505";  // 参考 PostgreSql 的实现
        }

        return false;
    }


    public override CPQuery SetPagedQuery(CPQuery query, int skip, int take)
    {
        return StdClientProvider.SetPagedQuery(query, skip, take);
    }

    public override Page2Query GetPagedCommand(BaseCommand query, PagingInfo pagingInfo)
    {
        return StdClientProvider.GetPagedCommand(query, pagingInfo);
    }

    public override string GetConnectionString(IDbConfig dbConfig, bool includeDatabase)
    {
        return PostgreSqlClientProvider.GetPostgreSQLConnectionString0(dbConfig, includeDatabase);
    }
}

初始化

在程序启动过程中，

ClownFishInit.InitDAL();

// 在 InitDAL 之后调用
KingbaseESClientProvider.RegisterProvider();

配置连接

<configuration>
    <connectionStrings>
        <add name="kingbase2" providerName="Kdbndp"
            connectionString="Host=192.168.1.1;database=mynorthwind;port=34321;Username=sa;Password=xxxxxxx"/>
    </connectionStrings>

    <dbConfigs>
        <add name="kingbase3" dbType="7777" server="192.168.1.1" port="34321" database="mynorthwind" uid="sa" pwd="xxxxxxx" args="" />
    </dbConfigs>
</configuration>

说明：

• 连接名“kingbase2”，设置了 providerName="Kdbndp"
  • 这是调用 DbClientFactory.RegisterProvider方法的第一个参数
• 连接名“kingbase3”，设置了 dbType="7777"
  • 它是 KingbaseESClientProvider.DatabaseType 的返回结果

测试代码

using( DbContext db = DbContext.Create("kingbase2") ) {

    // 确认此DbContext实例将会使用 KingbaseESClientProvider
    Assert.AreEqual(KingbaseESClientProvider.DatabaseTypeKingbaseES, db.DatabaseType);
    Assert.AreEqual(KingbaseESClientProvider.ProviderName, db.ProviderName);        
    
    long id = db.CPQuery.Create("select max(productid) from products").ExecuteScalar<long>();
    Assert.IsTrue(id > 0);
}


using( DbContext db = DbContext.Create("kingbase3") ) {

    Assert.AreEqual(KingbaseESClientProvider.DatabaseTypeKingbaseES, db.DatabaseType);
    Assert.AreEqual(KingbaseESClientProvider.ProviderName, db.ProviderName);

    long id = db.CPQuery.Create("select max(productid) from products").ExecuteScalar<long>();
    Assert.IsTrue(id > 0);
}

同时支持多种数据库

本文介绍在一个项目中同时支持多种数据库的实现方法

可以细分为2种场景：

• 场景1：同时支持 A库是SQLSERVER，B库是MYSQL，C库是PostgreSQL ，....
• 场景2：一套代码同时支持多种数据库，由部署时决定使用哪种数据库

实现思路

1，无论使用哪种数据库，都由数据库连接的配置来决定（示例中的2种配置方式可任选一种），例如：

<connectionStrings>
    <add name="sqlserver" providerName="System.Data.SqlClient"
            connectionString="server=192.168.1.1;database=MyNorthwind;uid=user1;pwd=xxx"/>

    <add name="mysql" providerName="MySql.Data.MySqlClient"
            connectionString="server=192.168.1.1;database=MyNorthwind;uid=user1;pwd=xxx"/>

    <add name="postgresql" providerName="Npgsql"
        connectionString="Host=192.168.1.1;database=mynorthwind;port=15432;Username=postgres;Password=xxx"/>
</connectionStrings>

<dbConfigs>
    <add name="s1" dbType="SQLSERVER" server="192.168.1.1" database="MyNorthwind" uid="user1" pwd="xxx" args="" />
    <add name="m1" dbType="MySQL" server="192.168.1.1" database="MyNorthwind" uid="user1" pwd="xxx" args="" />
    <add name="pg1" dbType="PostgreSQL" server="192.168.1.1" port="15432" database="mynorthwind" uid="postgres" pwd="xxx" args="" />
</dbConfigs>

• 对于【场景1】，不同的【连接名称】表示要连接不同的数据库
• 对于【场景2】，【连接名称】应该是固定的，只要在部署时切换 providerName 或者 dbType 就可以了

2，然后在代码中只使用【连接名称】

例如：

using( DbContext db = DbContext.Create("mysql") ) {
    //...... CURD 代码
}

using( DbContext db = DbContext.Create("pg1") ) {
    //...... CURD 代码
}

ClownFish的数据库操作相关类型不区分数据库，所以只要是标准的SQL，就能支持所有的数据库。

编写特定的数据库代码

有些数据库有特殊的命令， 【标准SQL】无法支持，只能编写【特定的数据库代码】，那么可参考下面示例来解决：

using( DbContext db = DbContext.Create("连接名称") ) {
    
    if( db.DatabaseType == DatabaseType.PostgreSQL ){
        // 在这里编写特定的数据库代码
    }
    else if( db.DatabaseType == DatabaseType.DaMeng ){
        // 在这里编写特定的数据库代码
    }
    else{
        // 其它数据库可通用的代码
    }

    // 或者使用 ProviderName 来区分数据库种类
    if( db.ProviderName == "Kdbndp" ){
        // 在这里编写特定的数据库代码
    }
    else{
        // 其它数据库可通用的代码
    }
}

使用 XmlCommand

如果项目中使用XmlCommand访问数据库，那么同时支持多种数据库会比较简单。

首先，配置一个 本地参数 ：

ClownFish_XmlCommand_SupportMulitDbType=1

然后，只需要对特定数据库配置不同的XmlCommand即可，例如：

<XmlCommand Name="RandGetCustomer">
    <CommandText><![CDATA[select * from Customers limit 1 ]]></CommandText>
</XmlCommand>
<XmlCommand Name="RandGetCustomer.SQLSERVER">
    <CommandText><![CDATA[select top 1 * from Customers ]]></CommandText>
</XmlCommand>

这里定义了2个XmlCommand，其中

• "RandGetCustomer.SQLSERVER" 是专门针对 SQLSERVER 定制的
• "RandGetCustomer" 用于除SQLSERVER之外的所有数据库

针对特定数据库的 XmlCommand 命名规则：name.{DatabaseType}

其中 DatabaseType 是由 ClownFish 定义的一个枚举类型，
如果在 自行扩展BaseClientProvider 的类型中使用了DatabaseType不存在的名称，那么请使用对应的数字，例如："RandGetCustomer.7777"

示例调用代码

string dbConnName = GetConnectionName();

using( DbContext db = DbContext.Create(dbConnName) ) {

    var customer = db.XmlCommand.Create("RandGetCustomer").ToSingle<Customer>();
}

上面这段代码在运行时，会根据连接的数据库类别自动选择合适的XmlCommand，当发现：

• 数据库是 SQLSERVER 时，会选择 "RandGetCustomer.SQLSERVER"
• 其它数据库类别时，会选择 "RandGetCustomer"

请点击目录中的节点查看具体文档

创建第一个项目

打开 Visual Studio (最新版本) 创建一个 ASP.NET Core web 应用，如下图所示：

添加 nuget 包

按下图所示添加必要的包，请选择最新版本

调整代码

• 删除 Startup.cs
• 按下面方式调整 Program.cs

[assembly: Microsoft.AspNetCore.Mvc.ApiController]

namespace YourProjectNameSpace;

public class Program
{
    public static void Main(string[] args)
    {
        AppStartup.RunAsPublicServices("ApplicationName", args);
    }
}

注意下面这行代码，

[assembly: Microsoft.AspNetCore.Mvc.ApiController]

有了它，写Action会简单些，建议在Program.cs添加。

添加配置文件

ClownFish.App.config通常是一个项目必需的配置文件，
可从Nebula的demo项目中复制到你的新项目中，然后再调整，
也可以按照以下内容来创建（注意文件的编码方式 utf-8 with BOM）

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
    <appSettings>

        <!-- 配置服务的访问地址 -->
        <add key="configServiceUrl" value="http://hostxxx:8503" />

    </appSettings>    
</configuration>

注意：configServiceUrl 参数是每个微服务项目必需要的，用于指定配置服务的连接地址。

项目设置

请参考以下截图

或者直接修改 launchSettings.json 文件，可参考以下代码

{
  "profiles": {
    "Nebula.Venus": {
        "commandName": "Project",
        "environmentVariables": {
            "ASPNETCORE_ENVIRONMENT": "Development"
        },
        "applicationUrl": "http://0.0.0.0:8208"
    }
  }
}

开发服务接口

按照分层开发原则，一个服务接口的开发过程有2块：

• 开发一个 HttpAction 接口方法
• 开发对应的 BLL 实现方法

Controller/BLL 的职责可参考：分层开发职责

HttpAction 接口方法

示例代码：

[Route("/v20/api/WebSiteApp/Customer/")]
public class CustomerController : BaseController
{
    private CustomerBLL customerBLL { get; set; }
            
    
    [HttpGet]
    [Route("findByTel.aspx")]
    public async Task<List<Customer>> FindByTel(string tel)
    {
        using( DbContext dbContext = CreateTenantConnection() ) {
            return await customerBLL.FindByTel(tel);
        }
    }

说明：

• Controller继承于BaseController
• Controller/Action 都必须标记[Route(xxxx)]
• 每个Action方法必须显式标记[HttpGet] or [HttpPost]这类限定调用标记

BLL 实现方法

示例代码：

public class CustomerBLL : BaseBLL
{
    public async Task<List<Customer>> FindByTel(string tel)
    {
        var query = from x in dbContext.Entity.Query<Customer>()
                    where x.Tel.Contains(tel)
                    select x;

        return await query.ToListAsync();
    }

说明：

• BLL类继承于BaseBLL
• BLL方法不要打开数据库连接

分层开发，2个重要的基类

基于Nebula开发的项目应该尊守必要的 分层开发职责

本文主要介绍2个重要基类的核心功能。

BaseController核心功能

• 用户身份识别
  • bool IsAuthenticated 属性
    • 判断当前请求是否为已登录用户
  • IUserInfo GetUserInfo()
    • 获取当前请求的用户身份信息
  • string GetTenantId()
    • 用于获取【当前用户】所属于的租户ID
    • 如果是未登录用户，则会抛出异常
• 数据库连接管理
  • DbContext CreateMasterConnection()
    • 切换到主库
  • DbContext CreateAppDbConnection(string name)
    • 切换到应用库
  • DbContext CreateTenantConnection(string tenantId = null)
    • 切换到租户库，参数为null时调用GetTenantId()
• 服务之间调用
  • Task<T> SendRequestAsync<T>(HttpOption httpOption)
    • 发送HTTP请求，并返回结果
  • Task TransferRequestAsync(string url, string method = null)
    • 将当前请求转发给一个内部服务来处理
• HttpRequest/HttpResponse 方法封装
  • string GetHeader(string name)
  • void SetHeader(string name, string value)
  • string GetCookie(string name)
  • void SetCookie(string name, string value, TimeSpan? expiry = null)

BaseBLL核心功能

• 用户身份识别
  • IUserInfo GetUserInfo()
  • string GetTenantId()
• 数据库连接获取
  • DbContext tntContext 属性
    • 获取当前Controller中的【最近】打开的【租户库】连接对象
  • DbContext masterContext 属性
    • 获取当前Controller中的【最近】打开的【主库】连接对象
• 通用 Insert/Update/Delete
  • Task<long> InsertAsync<T>(T entity, DbContext dbContext)
  • Task<int> UpdateAsync<T>(T entity, DbContext dbContext)
  • Task<int> DeleteAsync<T>(object key, DbContext dbContext)
  • Task<T> GetByKeyAsync<T>(object key, DbContext dbContext)

服务调用

本文所说的服务调用是指微服务之间的相互调用，它们采用 HTTP 传输协议。

示例代码：

private static readonly string s_userServiceUrl = Settings.GetSetting("UserService_Url", true);

public async Task<MysmUserInfo> GetUserLoginInfo()
{
    // 获取当前用户ID
    WebUserInfo userInfo = (WebUserInfo)this.NHttpContext.GetUserInfo();

    // 构造HTTP请求参数
    HttpOption httpOption = new HttpOption
    {
        Method = "GET",
        Url = s_userServiceUrl + "/v20/api/userservice/user/detail.svc",
        Data = new { userInfo.UserId },
        Format = SerializeFormat.Json
    };

    // 调用 UserService
    PUser pUser = await this.SendRequestAsync<PUser>(httpOption);

    // .................
}

简单来说，其实就2个步骤：

1. 构造HttpOption对象，它表示一个HTTP请求参数
2. SendRequestAsync(httpOption); 发起远程调用并获取远程服务的返回结果

补充说明：

• HttpOption 还有3组(同步/异步)扩展方法：Send, GetResult, GetResult<T>，在微服务相互调用时，不要使用它们
• SendRequestAsync 方法与上面3组方法相比，还会在调用时传递以下数据
  • 用户身份凭证（Token）
  • x-nebula- 开头的请求头
  • Cookie
    • 要求设置参数：Nebula_SendRequestWithCopyCookie = 1

用户登录

一个完整的用户登录包括以下步骤：

1. 前端显示登录界面
2. 前端提交用户登录数据
3. 服务端接收用户登录数据
4. 服务端校验用户名和密码
5. 服务端生成登录凭证并发送登录凭证给客户端

服务端接收用户登录数据

需要开发一个Action，例如：

[LoginAction]
[HttpPost]
[Route("login.aspx")]
public string Login([FromForm]string userId, [FromForm]string password, [FromForm]string tenantId)

说明：

• 本示例使用 表单 数据格式
• 也可以用JSON方式提交

生成登录凭证并发送登录凭证给客户端

在生成登录凭证之前，我们需要创建一个用户信息对象来保存当前用户相关信息。
然后用这个用户对象生成 JWT Token，也就是登录凭证。
示例代码如下：

// 创建用户信息对象
IUserInfo userInfo = new WebUserInfo {
    TenantId = tenantId,
    UserId = userId,
    UserName = "Fish Li",
    UserRole = "Admin"
};

// 定义凭证过期时间
int seconds = 1200;

// 生成登录凭证并发送登录凭证给客户端
AuthenticationManager.Login(userInfo, seconds);

AuthenticationManager.Login方法内部有2个过程

• 先生成登录凭证
• 以COOKIE形式发送登录凭证给客户端

单独生成登录凭证

如果仅仅需要生成登录凭证，可以参考下面代码：

IUserInfo userInfo = new WebUserInfo {
    TenantId = tenantId,
    UserId = userId,
    UserName = "Fish Li",
    UserRole = "Admin"
};

int seconds = 1200;
string token = AuthenticationManager.GetLoginToken(userInfo, seconds);

用户信息对象

Nebula中定义了 IUserInfo 接口用来表示用户信息，
同时提供了2个实现类可供使用：

• WebUserInfo
• AppClientInfo

如果不满足需求，请自行实现接口。

登录凭证续期

Nebula支持2种方式从前端到服务端传递登录凭证：

• Cookie模式
• 请求头模式

无论采用哪种模式，都可以支持登录凭证续期：

• Cookie模式：会产生一个新的Cookie，浏览器会自动接收，无需开发处理。
• 请求头模式：会产生一个【响应头】，此时需要客户端识别并保存，然后在后续请求中使用它。

身份认证

词汇解释

• Authenticate：身份认证，识别用户身份的过程。
• Authorize：授权检查，确定当前用户是否有权访问资源的过程。

工作机制

• Authenticate，由框架实现，对所有请求都有效（一直开启）
  • 它由一个专门的HttpModule来实现，当发现有用户登录身份凭证时（Cookie或者请求头），会自动识别
  • 会自动续期用户的登录身份凭证
  • 用户登录身份凭证可以微服务之间透明传递，例如 A->B->C 的服务调链，服务C仍然能获取用户身份并实现安全校验。
  • 用户识别的过程发生在进入 HttpAction 之前，所以可在 HttpAction 中获取当前用户信息
    
    
• Authorize，框架提供必要（基础的）验证手段 [Authorize] 来实现
  • [Authorize] 可以标记在 Controller 或者 Action 上面，以表达不同的应用范围
  • [Authorize] 支持4类过滤方式：用户名, 角色，权限号，用户身份类型IUserInfo
  • [AllowAnonymous] 可以表示允许【匿名用户】访问

实现过程

当用户成功登录后，服务端会生成一个登录凭证（JWT Token），
只要后续请求带上这个登录凭证，服务端就可以知道当前是谁，即还原 UserInfo 对象，
这个过程是在 Nebula中自动完成的，包含凭证继期处理，开发人员不需要参与处理。

身份认证 结束后，就可以在 HttpModule/Controller 中获取用户信息了。

授权检查

词汇解释

• Authenticate：身份认证，识别用户身份的过程。
• Authorize：授权检查，确定当前用户是否有权访问资源的过程。

仅允许已登录用户访问

Nebula默认支持在Controller/Action上采用标记的方式做授权申明，例如：

[HttpPost]
[Authorize]  // 注意这里
[Route("aaa.aspx")]
public string SomeMethod()

[Authorize] 标记打在Action方法上，表示当前方法仅支持【已登录用户】访问。

也可以将 [Authorize] 标记在Controller类型上，例如：

[Authorize]  // 注意这里
[Route("/v20/api/WebSiteApp/Customer/")]
public class CustomerController : BaseTestDatabaseController

此时，【等同于】将 [Authorize] 标记在所有Action方法上。
如果此时需要排除某个方法，可以在Action上标记 [AllowAnonymous]

[HttpPost]
[AllowAnonymous]  // 注意这里
[Route("bbb.aspx")]
public string SomeMethod2()

小结：

• [Authorize] 用于显式描述 当前功能仅限【已登录用户】访问
• [AllowAnonymous] 用于显式描述 当前功能允许供【未 登录用户】访问
• 以上2个标记可以出现在Controller/Action上，Action的标记将优先查找

注意：

• [Authorize], [AllowAnonymous] 这2个修饰属性定义在 Nebula.Common 命名空间下面。

基于角色的授权

以下标记表示当前资源仅允许 Admin， RoleX 这2类角色访问

[Authorize(Roles = "Admin,RoleX")]

当前用户是什么角色在创建 IUserInfo 时指定，例如：

IUserInfo userInfo = new WebUserInfo {
    TenantId = tenantId,
    UserId = userId,
    UserName = "Fish Li",
    UserRole = "Admin"
};

基于用户身份类型的授权

[Authorize(UserInfoType = typeof(VenusUserInfo))]

// 或者验证第三方客户端
[Authorize(UserInfoType = typeof(AppClientInfo))]
[Authorize(UserInfoType = typeof(AppClientInfo), Users = "AppId_xxxx")]

对应的登录代码片段：

VenusUserInfo userInfo = new VenusUserInfo {
    UserId = user.Name,
    UserName = user.Value,
};

int seconds = 7 * 24 * 60 * 60; // 7 day
AuthenticationManager.Login(userInfo, seconds);

简单理解:

• 在调用 AuthenticationManager.Login(...) 时使用了某个特定的【用户身份类型】
• 在授权检查时，就要求必须与登录时指定的【用户身份类型】一致

基于特定用户的授权

如果是比较简单的小项目，可以采用直接用户名的方式授权，例如：

[Authorize(Users = "Name1,Name2")]

• "Name1,Name2" 就是允许的登录名，
• 也可以是一个，
• 多个名称用分号分开

基于权限编号的授权

比较复杂的应用程序，可以按功能模块来实现授权，例如：

[Authorize(Rights = "A002")]

"A00002" 表示一个功能权限编号。

使用这个授权需要实现一个 BaseUserAuthProvider 抽象类，例如：

public class MyUserAuthProvider : BaseUserAuthProvider
{
    private IUserInfo _userInfo;

    protected override void SetDbContext(IUserInfo userInfo)
    {
        _userInfo = userInfo;

        // 注意：FindRole，FindUser 的默认实现方式是需要查询数据库的，
        // 因此需要在这里切换数据库连接，即设置基类的 _dbContext 字段
    }

    public override RightsRole FindRole(string roleId)
    {
        // 此方法用于获取角色对应的权限描述，这里不演示
        return null;
    }

    public override RightsUser FindUser(string userId)
    {
        // 这里是演示代码，就直接返回了，真实情况需要查询数据库或者缓存服务
        return new RightsUser {
            UserId = userId,
            UserName = _userInfo.UserName,
            RoleIds = "Admin",
            AllowRights = "22,23"
        };
    }
}

重点解释：

• FindUser方法，返回某个用户直接允许的权限，
• FindRole方法，返回某个用户所属的角色，
• 角色也可以赋予权限，
• 还可以给用户赋予否决权限，用于排除继承权限，
• 所以最终用户的权限会合并：单独授权 + 角色继承 - 否决权限

注册 UserAuthProvider

public class Program
{
    public static void Main(string[] args)
    {
        AppStartOption startOption = new AppStartOption {

            // 自定义的授权提供者
            UserAuthProviderType = typeof(MyUserAuthProvider)
        };
        
        AppStartup.RunAsWebsite("XDemo.WebSiteApp", args, startOption);
    }        
}

在代码中判断用户和权限

可以调用 Controller.CheckRights("...") ，例如：

public class ActionBLL : BaseBLL
{
    private void CheckOwner(ApiAction existObject)
    {
        IUserInfo user = this.GetUserInfo();

        bool currentIsAdmin = this.Controller.CheckRights("d001");
        string currentUser = user.UserId;


        // 只有管理员可以修改其他用户创建的接口
        // 普通用户不允许修改其他用户创建的接口
        if( currentIsAdmin == false && existObject.Createor != currentUser )
            throw new ForbiddenException("不能修改其他用户创建的接口。");
    }

扩展[Authorize]标记

如果觉得[Authorize]的功能不能满足业务需求，还可以从AuthorizeAttribute继承实现自己的派生类，只要重写AuthenticateRequest方法就可以了，以下代码是AuthorizeAttribute的默认实现：

/// <summary>
/// 执行授权检查验证的标记
/// </summary>
[AttributeUsage(AttributeTargets.Class | AttributeTargets.Method, AllowMultiple = false, Inherited = false)]
public class AuthorizeAttribute : Attribute
{
    /// <summary>
	/// 执行授权检查
	/// </summary>
	/// <param name="context"></param>
	/// <returns></returns>
	public virtual bool AuthenticateRequest(NHttpContext context)
    {
        IUserInfo userInfo = context.GetUserInfo();

        // 当前用户没有登录
        if( userInfo == null )
            return false;

        // 检查 IUserInfo 的实例类型是不是预期类型
        if( _userInfoType != null && userInfo.GetType() != _userInfoType )
            return false;

        // 没有任何明确的授权要求，此时只要是“已登录”用户就算通过检查
        if( _users == null && _roles == null && _rightsArray == null )
            return true;


        // 三种授权条件，只要一个符合就认为是检查通过
        if( CheckUser(context, userInfo) || CheckRole(context, userInfo) || CheckRights(context, userInfo) )
            return true;


        // 所有条件都不匹配，授权检查失败！
        return false;
    }

统一的认证与授权

现阶段，前后端分离已是一种非常流行的开发方式，后端以WebApi方式对外提供数据。
对于后端服务而言，它的调用方其实有3大类：

• 运行在浏览器的前端JS代码
• 集群内的其它服务
• 第三方应用程序

对应的调用方的身份有2大类：

• 真实的用户（人）
• 代码调用的HttpClient（对应后面2类）

在微服务架构下，服务的调用链会非常复杂，比如：

那么，这种场景下的 身份认证 和 授权检查 该如何实现呢？

Nebula提供了一种 统一的认证与授权 方案，用来解决这2个问题：

• 不同形式的客户调（调用方）
• 微服务的复杂调用链

具体包含以下方面：

• 不论何种调用方（客户端），统一用 IUserInfo 来表示身份，最后封装成 JWT-Token
  • JWT-Token可以通过3种方式传递
    • Cookie
    • 自定义请求头，例如：x-token: xxxxxxxxxxxxxxxxxx
    • 标准请求头，例如：Authorization: Bearer xxxxxxxxxxxxxxxxxx
• 微服务相互调用时，自动传递 JWT-Token
  • 因此每个服务都可以知道当前的用户身份
• 在服务端的Controller/Action上，使用 [Authorize(....)] 来执行授权检查
  • 没有[Authorize(....)]标记表示不做授权检查

实现过程

大致分类3个步骤：

• 创建登录表2个：用户表 和 客户端应用表
  • 用户表：每行记录表示一个现实世界中的用户，有登录名，用户名，密码，角色... 这些字段
  • 客户端应用表：每行记录表示一个 第三方 客户端，通常包含 AppName, AppId, LoginKey... 这些字段
• 登录检查时
  • 普通用户：登录名和密码验证通过后，创建一个 WebUserInfo 对象
  • 第三方客户端：AppId/LoginKey验证通过后，创建一个 AppClientInfo 对象
  • 然后调用 AuthenticationManager.Login 方法来生成登录凭证（JWT-Token）
• Controller/Action的授权检查
  • 请参考 授权检查

使用缓存

缓存分类

缓存无处不在，例如：CPU内核缓存，硬盘缓存，操作系统缓存，数据库缓存，等等……

以上这些缓存不受我们控制，本文只谈我们能在开发过程中能做到的缓存设计。

它们大致可分为以下几类

• 数据库结构缓存
  • 冗余表
  • 冗余字段
• 进程外缓存
  • 例如Redis
• 进程内缓存
  • 静态变量
  • 局部变量

冗余表/字段

有些时候为了优化查询性能，增加一些冗余的设计是很有必要的。

冗余表也称为“清洗表”，它需要结合后台任务来实现的，它通过定时快照方式将复杂的查询转成简单的结果表，让应用程序可以快速获取查询结果。

冗余字段就更简单了，它可以减少关联查询或者子查询的使用场景，简化查询，提升性能。

进程外缓存

将一些复杂的查询结果或者计算结果 放在独立的缓存服务中，本质上也是一种冗余的设计，通过冗余结果来简化获取过程，最终获得性能的提升。

最常见用法是缓存数据库查询结果，减少数据库的查询压力。

补充说明：

• 进程外缓存不一定性能非常好，它的价值是 减少数据库的压力
• 相比较进程内缓存，它可以实现在多个进程间共享缓存结果
• 数据库通常难以实现水平扩展，而缓存服务则比较容易

局部变量

局部变量（或者私有成员）的作用是减少一些反复的计算任务，例如：重复读取某个配置参数，重复查找集合，等等。

它的作用范围比较小，可以是一个方法内部，也可以是一个实例的生存周期内。

使用场景：

• 反复查找
  • 集合/字典
  • DOM
• 反复计算（封装）
  • 调用方法求值
  • 查询数据库求值
  • 调用服务求值
  • 获取配置
  • 创建对象
• 反复打开句柄
  • 数据库连接
  • Socket连接
  • 文件

静态变量

静态变量的生命周期和应用程序的生命周期一样长，可以长时间缓存结果，会有非常好的性能。

做为对比，从Redis获取结果会经过以下步骤：

• 对象序列化（Set方法）
• Redis ClientAPI 发起请求
• 网络数据传输
• Redis服务端处理
• 网络数据传输
• Redis ClientAPI 接收响应
• 反序列化

如果使用静态变量做缓存，以上这些步骤都没有了，显然性能可以得到 极大 的提升，

因此，从静态变量中获取结果相比于Redis服务会非常快，反之 Redis其实非常慢！

在程序中使用静态变量需要考虑3个问题

• 线程并发
• 缓存时间
• 内存占用

注意：如果不理解这几个问题，以及不知道解决方法，请不要使用静态变量！！

缓存工具类

为了简化使用过程，ClownFish提供了3个类型可供使用：

• CacheItem<T>
• CacheDictionary<T>
• AppCache

CacheItem用法示例

/// <summary>
/// 缓存允许登录的用户清单
/// </summary>
private static CacheItem<List<LoginUser>> s_users = new CacheItem<List<LoginUser>>(null, DateTime.MinValue, true);


private async Task<LoginUser> FindUser(string loginName)
{
    List<LoginUser> list = s_users.Get();
    if( list == null ) {
        list = await List();
        s_users.Set(list, DateTime.Now.AddSeconds(UserCache.CacheSeconds));
    }

    return list.FirstOrDefault(x => x.LoginName.Is(loginName));
}

CacheDictionary用法示例

/// <summary>
/// 租户相关的缓存工具类
/// </summary>
public static class TenantCache
{
    private static readonly CacheDictionary<string> s_cache = new CacheDictionary<string>();

    /// <summary>
    /// 根据请求头中的 x-custid 获取对应的租户ID
    /// </summary>
    /// <param name="controller">BaseController实例</param>
    /// <returns>租户ID</returns>
    public static async Task<string> GetTenanatId( BaseController controller)
    {
        if( controller == null )
            throw new ArgumentNullException(nameof(controller));

        string customerId = controller.GetHeader("x-custid");
        if( customerId.IsNullOrEmpty() )
            throw new ValidationException("当前请求中不包含 x-custid 请求头。");


        // 首先从缓存中获取
        string tenantId = s_cache.Get(customerId);

        if( tenantId.IsNullOrEmpty() ) {

            // 缓存中没有，就查询数据库
            using( DbContext dbContext = controller.OpenMasterConnection() ) {

                string sql = "select tenant_id from p_tenant where code = @code";
                var args = new { code = tenantId };
                tenantId = await dbContext.CPQuery.Create(sql, args).ExecuteScalarAsync<string>();

                if( tenantId.IsNullOrEmpty() )
                    return null;
            }

            // 将数据库中查询的结果放入缓存
            s_cache.Set(tenantId, tenantId, DateTime.Now.AddYears(1));
        }

        return tenantId;
    }
}

AppCache是一个静态类，
内部包含了一个CacheDictionary<object>成员，提供了3个公开方法供调用

/// <summary>
/// 简单的缓存工具类，供应用程序所有业务代码共用
/// </summary>
public static class AppCache
{
    /// <summary>
    /// 尝试从缓存中获取一个对象，如果缓存对象不存在，则调用“加载委托”进行加载并存入缓存。
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="key">缓存键</param>
    /// <param name="loadFunc">对象“加载委托”，用于当缓存对象不存时获取对象，调用结束后，新产生的对象将插入缓存。</param>
    /// <returns></returns>
    public static T GetObject<T>(string key, Func<T> loadFunc = null) where T : class

    /// <summary>
    /// 将一个对象添加到缓存中。
    /// </summary>
    /// <param name="key">缓存键</param>
    /// <param name="value">需要缓存的对象</param>
    /// <param name="expiration">缓存的过期时间</param>
    public static void SetObject(string key, object value, DateTime expiration)

    /// <summary>
    /// 删除指定键对应的缓存对象
    /// </summary>
    /// <param name="key">缓存键</param>
    public static void RemoveObject(string key)
}

缓存注意事项

对于SaaS应用程序来说，缓存一定要考虑多租户，因此缓存Key应该包含租户ID

请点击目录中的节点查看具体文档

消息管道模型

为了方便消息订阅和消息处理，ClownFish提供了一种称为消息管道的开发模型，如下图所示：

此模型有以下优点：

• 统一模型：支持一套消息处理代码，订阅不同的消息服务（消息来源），可理解为 与消息服务解耦。
• 功能增强：提供完善的日志和监控，支持重试处理，支持异常处理
• 简化代码：
  • 隐藏所有技术差异：
    • 订阅模式差异
    • 线程模型差异
    • 订阅者数量管理
    • 提交方式
    • 序列化等等细节
  • 消息处理只负责实现业务逻辑即可。
• 规范代码：将消息处理划分为多个阶段，避免代码风格迥异

消息订阅者

• 消息订阅者用于从消息队列服务中订阅（获取）消息，
• 订阅者接收到消息后，将消息发送到 消息管道，然后再交给 消息处理器 来处理
• 不同的队列来源 对应不同的订阅者（适配器模式）
• Nebula已内置了多种消息订阅者（适配器）
• 实际开发过程感知不到 消息订阅者 的存在，因为不需要与它交互

消息管道

消息处理管道的作用在于：

• 屏蔽各消息服务的差异
• 规范消息处理过程
• 支持错误重试
• 支持调用统计
• 支持性能监控
• 支持异常处理
• 产生日志（4类）

管道中的消息处理阶段（消息处理器提供了对应的虚方法）

• ValidateMessage
• ProcessMessage
• AfterProcess
• OnEnd
• OnError

消息处理器

无论使用哪种队列服务，所有消息最终是由 MessageHandler 来处理。

消息处理器

无论使用哪种队列服务，所有消息最终是由 MessageHandler 来处理。

特性说明及要求：

1. MessageHandler的实例随队列订阅者一直存在，直到进程退出。
2. 一个消息订阅者对应一个MessageHandler，开启多个订阅者可实现并行处理效果。
3. MessageHandler中所有数据成员在生命周期内持续可用，但由于实例隔离，会比静态成员更安全。
4. MessageHandler的实例化由框架来完成，自行实例化是没有意义的。
5. MessageHandler 不必关注消息是从哪里来，只需要处理它就可以了。
6. MessageHandler 响应消息管道的各个阶段，可以将复杂的逻辑拆分，有助于代码结构清晰易懂。
7. MessageHandler在执行时如果出现异常，默认会有异常处理，并且会启动重试机制，重新执行整个管道流程。
8. 由于消息管道的重试机制，以及消息未及时提交等等原因，消息有可能被重复发送到MessageHandler来处的，因此MessageHandler要实现 【幂等操作】

示例代码

/// <summary>
/// 消息处理器。
/// 它不关心消息从哪里来，只管处理特定类型的消息即可。
/// </summary>
public class PrintMessageHandler : BaseMessageHandler<Product>
{
    public override void ProcessMessage(PipelineContext<Product> context)
    {
        // 获取消息对象
        Product message = context.MessageData;

        string json = $"ProductID={message.ProductID},,{message.ProductName},,,{message.Remark}";
        string text = $"{_count.ToString().PadLeft(3, ' ')}; {json}";

        Console2.Info(text);
    }
}

小结：

• 必须从BaseMessageHandler<T>继承
• 类型参数T就是消息的数据类型
• 访问 context.MessageData 可获取消息对象

异常处理

BaseMessageHandler基类提供了3个与异常相关的方法用于定制行为（可以保持默认行为）：

• OnError：当消息处理过程中发生异常时会被调用。
  默认行为：不处理。
  说明：每条消息在处理过程中都会产生一条OprLog日志，如果有异常是会有记录的。
• IsNeedRetry：如果异常没有被处理，这个方法用于判断当前消息是否需要重试。
  默认行为：会根据异常类别做出判断。
• ProcessDeadMessage：在消息最终无法处理时调用。
  默认行为：将消息写入 temp\deadmsg 目录下，每个消息一个文件，即死消息文件。

死消息

消息管道默认有异常处理和重试机制，以下2种情况下消息会无法处理，形成死消息：

• 重试一直失败，达到最大次数
• 遇到的异常类型是不需要重试的异常，例如：ValidationException，Htt400

此时消息将被放弃重试，执行以下过程：

• 将消息写入 .\temp\deadmsg 目录下，每个消息一个文件。
• 向MQ服务端发送确认信号（表示消息已处理）
• 获取下一条消息，并继续处理

死消息文件示例

消息处理--同步

消息处理的开发过程有2个步骤：

• 开发 消息处理器
• 开启消息订阅

消息处理器--同步版本

示例代码（骨架）：

public class PrintMessageHandler : BaseMessageHandler<Product>
{
    public override void ProcessMessage(PipelineContext<Product> context)
    {
        //...具体的实现逻辑...
    }
}

开启消息订阅

注意：必须在程序初始时开户消息订阅

public static class AppInitializer
{
    public static void Init()
    {
        // 在这里填写 消息订阅 代码，可参考下面示例
    }
}

RabbitMQ版本

RabbitSubscriber.Start<Product, PrintMessageHandler>(new RabbitSubscriberArgs {
    SettingName = "连接参数名称XXXXX",
    QueueName = null,  // 队列名称是 Product类型的全名
    SubscriberCount = 2,  // 订阅者数量
    RetryCount = 5,    // 重试次数
    RetryWaitMilliseconds = 1000   // 重试的间隔时间
} );

Kafka版本

KafkaSubscriber.Start<Product, PrintMessageHandler>(new KafkaSubscriberArgs {
    SettingName = "连接参数名称XXXXX",
    GroupId = "消费组名称",
    Topic = null, // Topic 是 Product类型的全名
    SubscriberCount = 2,  // 订阅者数量
    RetryCount = 5,    // 重试次数
    RetryWaitMilliseconds = 1000   // 重试的间隔时间
} );

Redis版本

RedisSubscriber.Start<Product, PrintMessageHandler>(new RedisSubscriberArgs {
    SettingName = "连接参数名称XXXXX",
    Channel = null,   // null 表示使用Product的类型全名代替
    RetryCount = 5,    // 重试次数
    RetryWaitMilliseconds = 1000   // 重试的间隔时间
});

Pulsar版本

PulsarSubscriber.Start<Product, PrintMessageHandler>(new PulsarSubscriberArgs {
    SettingName = "连接参数名称XXXXX",
    Topic = "product",
    SubscriberCount = 2,  // 订阅者数量
    RetryCount = 5,    // 重试次数
    RetryWaitMilliseconds = 1000   // 重试的间隔时间
});

MMQ版本

// 创建一个内存队列实例，消息的生产方可以给它发送消息。
private static readonly MemoryMesssageQueue<Product> s_mmq = new MemoryMesssageQueue<Product>(MmqWorkMode.Sync);

internal static void StartSubscriber()
{
    MmqSubscriber.Start<Product, PrintMessageHandler>(new MmqSubscriberArgs<Product> {
        Queue = s_mmq,
        RetryCount = 0,    // 重试次数
        RetryWaitMilliseconds = 1000   // 重试的间隔时间
    });
}

小结：

• XxxxSubscriber.Start方法的第一个类型参数是：消息的类型
• 第二个类型参数是：MessageHandler类型
• 也就是：将什么类型的消息交给什么类型的处理器来处理
• XxxxSubscriber.Start方法都只有一个参数，XxxSubscriberArgs（请按F12查看它的成员）

注意事项

• 队列服务在使用前必须先在配置服务中注册，SettingName是连接的名称
• SubscriberCount是指订阅者数量，这个值 不是 越大越好！
• 最终全部订阅者数量 = SubscriberCount * 进程数量

不要硬编码 SubscriberCount

注意：SubscriberCount 这个参数，不要硬编码在代码中，应该使用一个参数来定义，例如：

RabbitSubscriber.Start<string, InvokeLogHandler3>(new RabbitSubscriberArgs {
    SettingName = ResNames.Rabbit,
    QueueName = typeof(InvokeLog).GetQueueName(),
    SubscriberCount = LocalSettings.GetUInt("Mercury_InvokeLog_RabbitSubscriber_Count", 3)
});

SubscriberCount 不应该太大

虽然增加 SubscriberCount 可以提高消息的吞吐量，但是订阅者也是会消耗资源的。

以RabbitMQ为例，每个订阅者都会创建一个TCP连接到RabbitMQ，而且还会有多个后台线程，因此它是一个比较消耗资源的对象，太多的连接也会给RabbitMQ制造一定的压力，所以建议这个参数不要太多，例如：对于一个队列来说，10个就够大了。

如果希望大幅提高吞吐量，可以参考下面示例：

private static void StartSubscribers()
{
    RabbitSubscriberArgs args = new RabbitSubscriberArgs
    {
        SettingName = RabbitSettings.DataBus_Rabbit,
        QueueName = "databus.request.new",
        SubscriberCount = 1,    // 只需要 1 个消息订阅者
    };

    RabbitSubscriber.StartAsync<NHttpRequest, HttpRequestMessageHandler>(args);
}


public class HttpRequestMessageHandler : AsyncBaseMessageHandler<NHttpRequest>
{
    private static readonly SemaphoreSlim s_semaphore;

    public override bool EnableLog => false;  // 不使用 MessageHandler 的日志

    static HttpRequestMessageHandler()
    {
        int concurrentCount = LocalSettings.GetUInt("DatabusMH_ConcurrentCount", 50);  // 控制并发量的参数
        s_semaphore = new SemaphoreSlim(concurrentCount);

        Console2.Info($"DatabusMH_ConcurrentCount = {concurrentCount}");
    }

    public override async Task ProcessMessage(PipelineContext<NHttpRequest> context)
    {
        NHttpRequest request = context.MessageData;

        await s_semaphore.WaitAsync();   // 这里控制并发

        // 异步执行发送请求的操作，当前方法中 【不等待】
        _ = ThreadUtils.RunAsync("HttpRequestMessageHandler_ProcessMessage", async () => {

            try {
                await SendRquest(request);
            }
            finally {
                s_semaphore.Release();
            }
        });
    }

    private static readonly bool s_logErrorRequest = LocalSettings.GetBool("DatabusMH_LogErrorRequest");

    private static async Task SendRquest(NHttpRequest request)
    {
        // 开启日志
        using( CodeSnippetContext ctx = new CodeSnippetContext(typeof(HttpRequestMessageHandler), "SendRequest", s_messageHandlerPerformance) ) {
            ctx.SetAsLongTask();

            try {
                int result = await SendRquest0(request, ctx.OprLog);

                // 如果非正常结束，就把请求记录下来
                if( s_logErrorRequest && result == 1 && ctx.OprLog.Request.IsNullOrEmpty() ) {
                    ctx.OprLog.Request = GetLogText(request);
                }
            }
            catch( Exception ex ) {
                ctx.SetException(ex);
            }
        }
    }

    // ...... 省略一些无关代码
}

示例小结：

• 采用 SemaphoreSlim 来控制并发数量，而不是增加消息订阅者数量
• 每条消息采用.NET线程池异步处理
• 由于HttpRequestMessageHandler.ProcessMessage 并没有实际处理消息，因此设置 EnableLog => false;
• 在消息处理代码中，using( CodeSnippetContext ctx = ...), 可记录消息的处理过程

消息处理--异步

消息处理的开发过程有2个步骤：

• 开发 消息处理器
• 开启消息订阅

消息处理器--异步版本

示例代码（骨架）：

public class PrintMessageHandler : AsyncBaseMessageHandler<Product>
{
    public override Task ProcessMessage(PipelineContext<Product> context)
    {
        //...具体的实现逻辑...
    }
}

启动消息订阅

注意：必须在程序初始时开户消息订阅

public static class AppInitializer
{
    public static void Init()
    {
        // 在这里填写 消息订阅 代码，可参考下面示例
    }
}

RabbitMQ版本

RabbitSubscriber.StartAsync<Product, PrintMessageHandler>(new RabbitSubscriberArgs {
    // 所有参数与同步版本一致，这里省略
} );

Kafka版本

KafkaSubscriber.StartAsync<Product, PrintMessageHandler>(new KafkaSubscriberArgs {
    // 所有参数与同步版本一致，这里省略
} );

Redis版本

RedisSubscriber.StartAsync<Product, PrintMessageHandler>(new RedisSubscriberArgs {
    // 所有参数与同步版本一致，这里省略
});

Pulsar版本

PulsarSubscriber.StartAsync<Product, PrintMessageHandler2>(new PulsarSubscriberArgs {
    // 所有参数与同步版本一致，这里省略
});

MMQ版本

// 创建一个支持异步操作的内存队列实例，消息的生产方可以给它发送消息。
private static readonly MemoryMesssageQueue<Product> s_mmq = new MemoryMesssageQueue<Product>(MmqWorkMode.Async);

internal static void StartSubscriber()
{
    MmqSubscriber.StartAsync<Product, PrintMessageHandler>(new MmqSubscriberArgs<Product> {
        // 所有参数与同步版本一致，这里省略
    });
}

小结：

• XxxxSubscriber.StartAsync方法的第一个类型参数是：消息的类型
• 第二个类型参数是：MessageHandler类型
• 也就是：将什么类型的消息交给什么类型的处理器来处理
• XxxxSubscriber.StartAsync方法都只有一个参数，XxxSubscriberArgs（请按F12查看它的成员）

注意事项

• 队列服务在使用前必须先在配置服务中注册，SettingName是连接的名称
• SubscriberCount是指订阅者数量，这个值 不是 越大越好！
• 最终全部订阅者数量 = SubscriberCount * 进程数量

RabbitMQ基础概念

本文将介绍 RabbitMQ 相关的开发基础，主要包含以下内容

1. RabbitMQ基础介绍
2. RabbitMQ工作模型
3. 交换机使用建议
4. 消息持久化
5. 死信队列
6. Virtual Hosts
7. 连接的生命周期
8. 高可用

RabbitMQ基础介绍

一个完整的RabbitMQ消息流转过程如下图：

其中包含以下对象：

• Producers消息生产者：负责产生消息，发送到RabbitMQ
• Exchanges交换机 ：接收消息，并投递到与之绑定的队列
• Queues消息队列：负责存储消息
• Consumers消息消费者：负责处理消息

这里需要注意

• 消息是发送到交换机，而不是队列
• 交换机与队列是独立工作的，它们之间存在一种映射关系（路由规则）
• 一条消息经过交换机，可以复制成多份，最终发送到不同的队列
• 一个队列可供多个订阅者订阅（并行消费消息）

Exchange
生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

Routing key
生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。 在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。 RabbitMQ为routing key设定的长度限制为255 bytes。

Binding
RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

Binding key
在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。 在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

Exchange Types
RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

fanout
fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

direct
direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。

topic
前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定： routing key为一个句点号“. ”分隔的字符串 （我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit” binding key与routing key一样也是句点号“. ”分隔的字符串 binding key中可以存在两种特殊字符“”与“#”，用于做模糊匹配，其中“”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）

headers
headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。 在绑定Queue与Exchange时指定一组键值对，当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对，如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。

RabbitMQ工作模型

• Broker: 接受客户端连接，实现AMQP实体服务。
• Vhost: 虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。
• Exchange: 消息交换机,它指定消息按什么规则,路由到哪个队列。
• Queue: 消息的载体,每个消息都会被投到一个或多个队列。
• Binding: 绑定，它的作用就是把exchange和queue按照路由规则绑定起来.
• Routing Key: 路由关键字,exchange根据这个关键字进行消息投递。
• Producer: 消息生产者,就是投递消息的程序.
• Consumer: 消息消费者,就是接受消息的程序.
• Connection: 客户端与某个broker的网络连接。
• Channel: 消息通道,在客户端的每个连接里,可建立多个channel.

交换机使用建议

• 对于大多数情况下，使用RabbitMQ自带的 amq.direct 就足够了，不需要自行创建交换机
• direct将也支持将一份消息发送到多个队列（复制），因此fanout并不是必要的
• 为了简化开发及排查过程，建议用消息的类型名称做为 bindingkey

消息持久化

队列和交换机有一个创建时候指定的标志durable, durable的唯一含义就是具有这个标志的队列和交换机会在重启之后重新建立,它不表示说在队列中的消息会在重启后恢复

消息持久化包括3部分

1. exchange持久化,在声明时指定durable => true hannel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);//声明消息队列，且为可持久化的

2. queue持久化,在声明时指定durable => true channel.QueueDeclare(QueueName, durable: true, exclusive: false, autoDelete: false, arguments: null);//声明消息队列，且为可持久化的

3. 消息持久化,在投递时指定delivery_mode => 2 channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());

如果exchange和queue都是持久化的,那么它们之间的binding也是持久化的,如果exchange和queue两者之间有一个持久化，一个非持久化,则不允许建立绑定.

注意：一旦创建了队列和交换机,就不能修改其标志了, 例如,创建了一个non-durable的队列,然后想把它改变成durable的,唯一的办法就是删除这个队列然后重现创建。

死信队列

死信队列介绍

• 死信队列：DLX，dead-letter-exchange
• 利用DLX，当消息在一个队列中变成死信 (dead message) 之后，它能被重新publish到另一个Exchange，这个Exchange就是DLX

消息变成死信有以下几种情况

• 消息被拒绝(basic.reject / basic.nack)，并且requeue = false
• 消息TTL过期
• 队列达到最大长度

死信处理过程

• DLX也是一个正常的Exchange，和一般的Exchange没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属性。
• 当这个队列中有死信时，RabbitMQ就会自动的将这个消息重新发布到设置的Exchange上去，进而被路由到另一个队列。
• 可以监听这个队列中的消息做相应的处理。

参考链接：https://www.jianshu.com/p/986ee5eb78bc

Virtual Hosts

Introduction
RabbitMQ is multi-tenant system: connections, exchanges, queues, bindings, user permissions, policies and some other things belong to virtual hosts, logical groups of entities. If you are familiar with virtual hosts in Apache or server blocks in Nginx, the idea is similar. There is, however, one important difference: virtual hosts in Apache are defined in the configuration file; that's not the case with RabbitMQ: virtual hosts are created and deleted using rabbitmqctl or HTTP API instead.

Logical and Physical Separation
Virtual hosts provide logical grouping and separation of resources. Separation of physical resources is not a goal of virtual hosts and should be considered an implementation detail.

For example, resource permissions in RabbitMQ are scoped per virtual host. A user doesn't have global permissions, only permissions in one or more virtual hosts. User tags can be considered global permissions but they are an exception to the rule.

Therefore when talking about user permissions it is very important to clarify what virtual host(s) they apply to.

小结：

• 如果应用程序使用的队列数量比较多，可以考虑采用这种方式来隔离，方便管理。
• 也可以用于隔离多套测试环境。

连接的生命周期

Connections are meant to be long-lived. The underlying protocol is designed and optimized for long running connections. That means that opening a new connection per operation, e.g. a message published, is unnecessary and strongly discouraged as it will introduce a lot of network roundtrips and overhead.

Channels are also meant to be long-lived but since many recoverable protocol errors will result in channel closure, channel lifespan could be shorter than that of its connection. Closing and opening new channels per operation is usually unnecessary but can be appropriate. When in doubt, consider reusing channels first.

Channel-level exceptions such as attempts to consume from a queue that does not exist will result in channel closure. A closed channel can no longer be used and will not receive any more events from the server (such as message deliveries). Channel-level exceptions will be logged by RabbitMQ and will initiate a shutdown sequence for the channel (see below).

连接应该是长久的。底层协议是为长时间运行的连接而设计和优化的。这意味着每次操作打开一个新连接， 例如发布一条消息，是不必要的，并且强烈反对这样做，因为这会引入大量的网络往返和开销。

通道也意味着寿命很长，但是由于许多可恢复的协议错误将导致通道关闭，通道的寿命可能比连接的寿命短。 每次操作关闭和打开新的通道通常是不必要的，但可以是适当的。当有疑问时，首先考虑重用通道。

通道级异常，比如试图从不存在的队列中消费，将导致通道关闭。关闭的通道将不再被使用， 并且将不再从服务器接收任何事件(如消息传递)。通道级异常会被RabbitMQ记录下来， 并且会启动通道的关闭序列(见下文)。

在Nebula中

• 订阅者使用的长连接，且支持中断后重新连接
• 生产者也支持长连接，用名称来区分维护，
• new RabbitClient时，如果不指定连接名（第2个参数），连接将在RabbitClient结束时释放。

高可用

RabbitMQ的高可用涉及以下方面：

• 安装部署
  • 集群复制：https://www.rabbitmq.com/clustering.html
• 参数配置
  • 镜像设置：https://www.rabbitmq.com/ha.html
  • 仲裁队列：https://www.rabbitmq.com/docs/quorum-queues
  • 队列策略：https://www.rabbitmq.com/parameters.html#policies
  • 持久化： 交换机，队列，消息
• 客户端
  • 连接自动恢复：https://www.rabbitmq.com/dotnet-api-guide.html#recovery
  • 消息确认：Message acknowledgment

ClownFish默认使用 classic queues，如果希望使用 quorum-queues ，可在配置服务中添加一个参数：

ClownFish_RabbitMQ_DefaultQueueType=quorum

强烈建议 一定要为队列设置 【队列策略】 来限制每个队列的最大长度，例如：

显示效果：

使用RabbitMQ

本文将介绍RabbitMQ的使用技巧，主要包含以下内容

1. 在配置服务中注册连接
2. 创建队列
3. 发送消息
4. 订阅消息
5. 消息的延迟(定时)处理
6. 多队列&多进程-自动配对订阅

参考链接

官方API参考： https://www.rabbitmq.com/dotnet-api-guide.html

官方教程： https://www.rabbitmq.com/tutorials/tutorial-one-dotnet.html

在配置服务中注册连接

在Nebula中使用RabbitMQ必须先在配置服务中注册RabbitMQ连接，
具体做法中在settings表中增加一条记录（也可通过AdminUI界面来操作），例如：

字段说明

• name: 连接名称，在代码中使用
• value: 连接参数，可用数据成员参考下面RabbitOption代码
• restype=1: 用于开启Venus监控

/// <summary>
/// RabbitMQ的连接信息
/// </summary>
public sealed class RabbitOption
{
    /// <summary>
    /// VHost，默认值："/"
    /// </summary>
    public string VHost { get; set; } = "/";
    
    /// <summary>
    /// RabbitMQ服务地址。【必填】
    /// </summary>
    public string Server { get; set; }
    
    /// <summary>
    /// 连接端口（大于 0 时有效）
    /// </summary>
    public int Port { get; set; }

    /// <summary>
    /// HTTP协议的连接端口（大于 0 时有效）
    /// </summary>
    public int HttpPort { get; set; }
    
    /// <summary>
    /// 登录名。【必填】
    /// </summary>
    public string Username { get; set; }
    
    /// <summary>
    /// 登录密码。
    /// </summary>
    public string Password { get; set; }
}

参数值配置示例

server=RabbitHost;username=fishli;password=aaaaaaaaaaaa;vhost=nbtest

创建队列

下面示例代码演示了2种创建队列的方法，
它们都使用了默认的 amq.direct 交换机。

public static class RabbitInit1
{
    public static readonly string SettingName = "Rabbit-DEMO";


    /// <summary>
    /// 创建【强类型】的消息队列
    /// </summary>
    private static void CreateQueue1()
    {
        using( RabbitClient client = new RabbitClient(SettingName) ) {

            // 队列名称，bindingKey 二者相同，都是 Product类型的全名
            client.CreateQueueBind(typeof(Product));
        }
    }
}

public static class RabbitInit2
{
    public static readonly string QueueName = "DEMO-QUEUE";

    public static readonly string SettingName = "Rabbit-DEMO";

    /// <summary>
    /// 创建【自由名称】的消息队列
    /// </summary>
    private static void CreateQueue2()
    {
        using( RabbitClient client = new RabbitClient(SettingName) ) {

            // 队列名称，bindingKey 二者相同，都是 "DEMO-QUEUE"
            client.CreateQueueBind(QueueName);
        }
    }
}

CreateQueueBind方法签名如下：

/// <summary>
/// 创建队列并绑定到交换机
/// </summary>
/// <param name="queue">队列名称</param>
/// <param name="exchange">交换机名称，默认值： "amq.direct"</param>
/// <param name="bindingKey">从交换机到队列的映射标记</param>
/// <param name="argument">调用QueueDeclare时传递的argument参数</param>
public virtual void CreateQueueBind(string queue, string exchange = null, string bindingKey = null, IDictionary<string, object> argument = null)

/// <summary>
/// 创建队列并绑定到交换机
/// </summary>
/// <param name="dataType">消息的数据类型，最终创建的队列名称就是消息数据类型的全名，bindingKey与队列同名</param>
/// <param name="exchange">交换机名称，默认值： "amq.direct"</param>
/// <param name="argument">调用QueueDeclare时传递的argument参数</param>
public void CreateQueueBind(Type dataType, string exchange = null, IDictionary<string, object> argument = null)

发送消息

可参考以下代码：

internal class RabbitProducer : BaseController
{
    /// <summary>
    /// 将数据发送到队列，【强类型】方式
    /// </summary>
    public void SendData1(Product product)  // 方法 1
    {
        // 如果只是     【偶尔】    需要发送消息，可以这样调用

        using( RabbitClient client = new RabbitClient(RabbitInit.SettingName) ) {

            // 往 RabbitMQ 发送一条消息，routingKey就是参数的类型全名
            client.SendMessage(product);
        }
    }

    public void SendData2(Product product)  // 方法 2
    {
        // 往 RabbitMQ 发送一条消息，routingKey就是参数的类型全名
        // 它会在整个应用程序内部维护一个共享连接，在多次调用时性能会更好。

        this.SendRabbitMessage(RabbitInit.SettingName, product);
    }

    public void SendData3(Product product)  // 方法 3
    {
        // RabbitClient构造方法的第二个参数是一个 “共享连接” 的名称，

        // 默认值是null，表示连接仅在 RabbitClient 内部使用，RabbitClient离开作用域后将关闭连接。

        // 如果指定了连接名称，那么连接将会保持，在下次构造RabbitClient时将会重用已存在的连接。
        // 共享连接的名称建议使用 nameof(类型名称) ，表示连接仅供某个类型共用。

        using( RabbitClient client = new RabbitClient(RabbitInit.SettingName, nameof(RabbitProducer)) ) {

            client.SendMessage(product);
        }
    }

    /// <summary>
    /// 将数据发送到队列，指定routingKey
    /// </summary>
    public void SendData4(Product product)
    {
        using( RabbitClient client = new RabbitClient(RabbitInit.SettingName, nameof(RabbitProducer)) ) {

            // 往 RabbitMQ 发送一条消息，routingKey由第3个参数指定
            client.SendMessage(product, null, RabbitInit.QueueName);
        }
    }

}

SendMessage方法签名如下：

/// <summary>
/// 往队列中发送一条消息。
/// </summary>
/// <param name="data">要发送的消息数据</param>
/// <param name="exchange">交换机名称，默认值： "amq.direct"</param>
/// <param name="routingKey">消息的路由键</param>
/// <param name="basicProperties"></param>
/// <returns>消息体长度</returns>
public int SendMessage(object data, string exchange = null, string routingKey = null, IBasicProperties basicProperties = null)

RabbitClient构造方法签名如下：

/// <summary>
/// 构造方法
/// </summary>
/// <param name="settingName">配置服务中的Rabbit连接名称</param>
/// <param name="connectionName">
/// 连接名称，
/// 如果不指定，表示连接在使用结束后关闭， 
/// 如果指定，那么连接将会一直打开，供后续同名的connectionName使用。</param>
public RabbitClient(string settingName, string connectionName = null)

订阅消息

可参考以下代码：

public class RabbitConsumer
{
    /// <summary>
    /// 订阅消息，【强类型】方式
    /// </summary>
    public static void Start1()
    {
        // 创建队列订阅者，并开启监听
        RabbitSubscriber.Start<Product, PrintMessageHandler>(new RabbitSubscriberArgs {
            SettingName = RabbitInit.SettingName,
            QueueName = null,  // 队列名称是 Product类型的全名
            SubscriberCount = LocalSettings.GetUInt("RabbitSubscriber.Count", 2)
        });
    }

    /// <summary>
    /// 订阅消息，【自由名称】方式
    /// </summary>
    public static void Start2()
    {
        // 创建队列订阅者，并开启监听
        RabbitSubscriber.Start<Product, PrintMessageHandler>(new RabbitSubscriberArgs {
            SettingName = RabbitInit.SettingName,
            QueueName = RabbitInit.QueueName,
            SubscriberCount = LocalSettings.GetUInt("RabbitSubscriber.Count", 2)
        });
    }
}

消息的延迟(定时)处理

实现思路：

• 创建一个队列 WaitRetry，为它设置 死信队列 ReSend
• 发送消息时指定过期时间，
• 消息进入队列后，由于没有订阅者来消息，会在过期后进入死信队列
• 订阅者订阅死信队列，获取到【延迟】的消息

注意：WaitRetry/ReSend 只是示例代码中的名称，并没有强制要求！

可参考下图：

示例代码：

private static void CreateRetryMQ()
{
    //  “等待重试” 队列
    string waitRetryQueue = Res.GetWaitRetryQueue();

    // “重试” 队列
    string resendQueue = Res.GetReSendQueue();

    using( RabbitClient client = Res.CreateRabbitClient() ) {

        // 为 waitRetryQueue 配置 死信队列
        Dictionary<string, object> args = new Dictionary<string, object>();
        args["x-dead-letter-exchange"] = Exchanges.Direct;
        args["x-dead-letter-routing-key"] = resendQueue;
        
        client.CreateQueueBind(waitRetryQueue, null, null, args);                
        client.CreateQueueBind(resendQueue);
    }

    // 创建队列的消息订阅者
    RabbitSubscriber.Start<MessageX, ReSendHandler>(new RabbitSubscriberArgs {
        SettingName = Res.RabbitSettingName,
        QueueName = resendQueue,
    });
}

此时可以在RabbitMQ的管理界面看到：

public void SendRetryMessage(MessageX message, int waitMilliseconds)
{
    using( RabbitClient client = Res.CreateRabbitClient() ) {

        // 注意：消息发送的目标队列 是没有订阅者的，
		// 消息会一直等到 “过期” 被移到 “死信队列”，在哪里才会被处理。
        IBasicProperties properties = client.GetBasicProperties();
        properties.Expiration = waitMilliseconds.ToString();

        string routing = Res.GetWaitRetryQueue();
        client.SendMessage(message, null, routing, properties);
    }
}

剩下的事情就是订阅 ReSend 队列并处理消息，这里忽略这个过程……

多队列&多进程-自动配对订阅

回顾一下 简单的队列使用场景
假如我们有一个数据类型 XMessage
我们可以为它创建一个队列，名字也叫 XMessage，
然后我们可以用下面的代码来订阅这个队列：

RabbitSubscriberArgs args = new RabbitSubscriberArgs{
     SettingName = "Rabbit连接的配置名称",
     QueueName = "XMessage",
     SubscriberCount = 3
};
RabbitSubscriber.Start<XMessage, XMessageHandler>(args);

如果我们将程序运行2个实例，那么对于 XMessage 队列，它将有6个订阅者（2个实例 * 3个订阅线程），
所有消息将会均匀分配到6个订阅者。

多队列&多进程--使用场景
假设：XMessage类型的消息，由于种种原因需要分裂成多个“相同的”队列，
例如按租户ID取模，之后需要将同一个租户的所有消息固定发送到与它对应的队列。
此时进程（容器）运行多个实例，如何保证每个队列只被一个进程订阅（尤其要考虑进程重启）？

多队列场景示例
XMessage队列需要创建10个，分别是 XMessage1, XMessage2,,,,,XMessage10
它们仍然接收 XMessage 这个类型的消息，只是按租户ID将它们分开而已，
也就是说：

• 如果XMessage消息中，tenantid = my596d74ca5eeb1，那么这条消息就投递到XMessage1队列，
• 如果XMessage消息中，tenantid = my596d74ca5eeb2，那么这条消息就投递到XMessage2队列，
• 如果XMessage消息中，tenantid = my596d74ca5eeb3，那么这条消息就投递到XMessage3队列，

此时我们的程序运行5个实例，那么订阅代码如何编写？
因为容器内的进程运行起来是一模一样的，不可能在代码或者配置中指定队列名称！

• 使用分布式锁，在锁范围内找到没有被订阅的队列，然后订阅它！

参考代码

/// <summary>
/// 多队列-多进程 自动配对订阅实现方案，
/// 假如有 10 个队列，5个进程，此时每个进程将订阅2个队列
/// </summary>
public static void Demo()
{
    // 此方案的大致思路是：
    // 1，先获取一个分布式锁，保证多个进程互斥（只有一个进程能进入执行）
    // 2，获取所有队列信息，检查将要订阅的队列是否有订阅者，如果队列没有订阅者，就订阅它


    // 创建一个分布式锁，锁将保证多个进程时，只有一个进程能进入执行
    using( GlobalLock locker = new GlobalLock("一个特殊的锁标识", TimeSpan.FromSeconds(30)) ) {

        RabbitMonitorClient client = new RabbitMonitorClient("rabbitmq-连接名称");

        // 获取所有队列信息，并过滤【特定】
        List<QueueInfo> list = client.GetQueues(10_000).Where(x => x.Name.StartsWith("队列名称前缀")).ToList();

        // 记录当前进程已订阅的队列数量
        int count = 0;

        foreach( var q in list ) {

            // 如果某个队列没有订阅者，就表示当前进程可以订阅它
            if( q.Consumers == 0 && count < 2 ) {

                // 订阅某个队列
                RabbitSubscriberArgs args = new RabbitSubscriberArgs {
                    SettingName = "Rabbit连接配置名称",
                    QueueName = q.Name,
                    SubscriberCount = 3
                };
                RabbitSubscriber.Start<XMessage, XMessageHanlder>(args);
                count++;
            }
        }
    }
}

RabbitMQ开发建议

主要包含以下内容

1. 配置RabbitMQ连接
2. 消息队列的创建
3. 发送消息
4. 发送消息使用建议
5. 订阅消息
6. 处理消息
7. 幂等操作
8. 一条消息多次使用
9. 订阅方性能优化
10. 指标监控

配置RabbitMQ连接

RabbitMQ在使用时，会有一些与连接相关的参数，例如，服务地址，端口，用户名，密码之类的参数。
为了安全，请将这些信息保存到配置服务中。
这个时候，我们需要给配置项取一个连接名称。

建议：“连接名称” 要能体现出属于哪个应用。

例如： "Nebula_Metis_Rabbit"

这样做有3个好处：

• 可以防止名称重复
• 便于识别连接的用途
• 便于以后实现队列迁移

消息队列的创建

基本概念介绍：

• 在消息订阅时，会要求指定要订阅哪个队列，因此队列必须要提前创建。
• 在消息发送时，消息发送到交换机就完事了，在RabbitMQ内部会有一个路由机制转发到队列，所以严格意义来说，发送方是不需要关心队列是否存在的。

建议：不管是发送方还是订阅方，在程序初始化时，都应该提前将队列创建好。

队列的创建方法：

public static class RabbitInit
{
    public static readonly string QueueName = "mq111111111";

    public static readonly string SettingName = "Nebula_Metis_Rabbit";


    /// <summary>
    /// 创建【强类型】的消息队列
    /// </summary>
    private static void CreateQueue1()
    {
        using( RabbitClient client = new RabbitClient(SettingName) ) {

            // 创建队列，队列名称是 Product类型的全名
            client.CreateQueueBind(typeof(Product));
        }
    }

    /// <summary>
    /// 创建【自由名称】的消息队列
    /// </summary>
    private static void CreateQueue2()
    {
        using( RabbitClient client = new RabbitClient(SettingName) ) {

            // 创建队列，队列名称由参数指定
            client.CreateQueueBind(QueueName);
        }
    }
}

注意：

• CreateQueueBind方法的传入参数

发送消息

/// <summary>
/// 将数据发送到队列，【强类型】方式
/// </summary>
public void SendData1(Product product)  // 方法 1
{
    // 如果只是偶尔需要发送消息，可以这样调用

    using( RabbitClient client = new RabbitClient(RabbitInit.SettingName) ) {

        // 往 RabbitMQ 发送一条消息，队列名称就是参数的类型全名
        client.SendMessage(product);
    }
}

public void SendData2(Product product)  // 方法 2
{
    // 往 RabbitMQ 发送一条消息，队列名称就是参数的类型全名
    // 它会在整个应用程序内部维护一个共享连接，在多次调用时性能会更好。

    this.SendRabbitMessage(RabbitInit.SettingName, product);
}

public void SendData3(Product product)  // 方法 3
{
    // RabbitClient构造方法的第二个参数是一个 “共享连接” 的名称，
    // 默认值是null，表示连接仅在 RabbitClient 内部使用，RabbitClient离开作用域后将关闭连接。

    // 如果指定了连接名称，那么连接将会保持，在下次构造RabbitClient时将会重用已存在的连接。
    // 共享连接的名称建议使用 nameof(类型名称) ，表示连接仅供某个类型共用。

    using( RabbitClient client = new RabbitClient(RabbitInit.SettingName, nameof(RabbitProducer)) ) {

        client.SendMessage(product);
    }
}

/// <summary>
/// 将数据发送到队列，【自由名称】方式
/// </summary>
public void SendData4(Product product)
{
    using( RabbitClient client = new RabbitClient(RabbitInit.SettingName, nameof(RabbitProducer)) ) {

        // 往 RabbitMQ 发送一条消息，队列名称由第3个参数指定
        client.SendMessage(product, null, RabbitInit.QueueName);
    }
}

以上4种方式都可行，根据注释来选择。

它们都只提供了二个必需参数：

• SettingName： 表示 RabbitMQ 在配置服务中的连接配置项
• product: 消息对象，发送时将根据变量的类型名称，找到对应的队列名称

发送消息使用建议

Nebula提供4种消息发送方式：

• 方法1：整体速度最差，主要慢在与RabbitMQ的连接上，用于偶尔发送一条消息的场景，每次调用结束后会关闭连接。
• 方法2：其实是“方法3”的包装版本，将“连接名称”固定了而已。
• 方法3：允许指定共享连接名称，如果程序有多个地方需要消息时，可以通过指定连接名称来减少访问冲突。
• 方法4：指定队列名称（其实是路由名称），用于“自由队列”

使用建议：

• 在程序初始化时，如果需要创建队列，使用“方法1” （用完就关闭的连接）
• 如果消息的发送量非常少，也推荐使用“方法1”
• 在Controller/Action中 推荐使用 “方法2”
• 其它地方可以使用“方法3”或者“方法4”，但是连接名不要乱取，每个名字意味着会产生一个长连接（永不关闭）

订阅消息

可参考以下代码：

/// <summary>
/// 订阅消息，【强类型】方式
/// </summary>
public static void Start1()
{
    // 创建队列订阅者，并开启监听
    RabbitSubscriber.Start<Product, PrintMessageHandler>(new RabbitSubscriberArgs {
        SettingName = RabbitInit.SettingName,
        QueueName = null,  // ### 注意这个参数：队列名称是 Product类型的全名
        SubscriberCount = LocalSettings.GetUInt("Product_RabbitSubscriber_Count", 1)
    });
}

/// <summary>
/// 订阅消息，【自由名称】方式
/// </summary>
public static void Start2()
{
    // 创建队列订阅者，并开启监听
    RabbitSubscriber.StartAsync<Product, PrintMessageHandler2>(new RabbitSubscriberArgs {
        SettingName = RabbitInit.SettingName,
        QueueName = "mq111111111",  //  ### 注意这个参数
        SubscriberCount = LocalSettings.GetUInt("Product_RabbitSubscriber_Count", 1)
    });
}

注意：订阅者的数量一定可配置！

说明：

• 每个订阅者不仅包含一个Rabbit连接，还包含了一个 “消息处理管道"
• 在 消息处理管道 中，还包含了与消息类型匹配的 MessageHandler
• 消息处理管道 内部有重试和异常处理机制，MessageHandler 可以不用关心

处理消息

请在项目中创建一个Messages目录，然后参考下图创建 MessageHandler

消息最终是由 MessageHandler来处理的。

幂等操作

幂等操作是什么，请自行搜索，这里不解释。

幂等操作常规实现方法（方法1）

• 为消息指定一个GUID成员，例如： public Guid MessageGuid {get; set;}
• 对应的数据库表中，为 MessageGuid 创建【唯一索引】
• 在消息处理时
  • 打开数据库事务
  • 调用 entity.Insert2() 扩展方法将消息入库，判断返回值：是否为重复插入，
    • 如果数据行已存在，可以调用 EndProcess() 来结束当前管道过程。
    • 否则按以下步骤执行
      • 处理消息（执行具体的业务过程）
      • 提交数据库事务

方法2
如果：

• 业务过程比较简单，例如：只有一个操作
• 或者业务操作过程不支持回退（取消）

那么可以不使用上面的方法， 而是直接在处理消息！
因为消息在处理结束后，ClownFish 会给RabbitMQ发送一个ACK信号，此时消息将会删除，也不会被重复处理。

消息的幂等操作由于涉及到消息队列这个外部服务，因此需要分布式事务来保证一致性，但是，
分布式事务本质上都不是完美的，上面二种做法都存在极小概率的缺陷，所有不必纠结！

一条消息多次使用

场景举例：
客户端上传了一条日志，在服务端需要做三个处理

1. 持久化到MySQL
2. 同时写一份到Elasticsearch
3. 实时数据分析

实现方法：

• 创建三个队列，设置相同的 Routing key ，消息将同时发送到三个队列，然后分开处理。

订阅方性能优化

对于消息的订阅方而言，提升消息的吞吐量有2种方法：

• 增加订阅者数量
• 优化MessageHandler的代码性能

订阅者数量不可能无限增加，所以优化代码性能是很有必要的，超出性能阀值会记录到性能日志中。

订阅者 数量如何确定？

1. 订阅者的数量不要太大，每个订阅者对应着4个后台线程(线程池)， 所以每增加1个订阅者，也会增加4个线程，太多无用的线程对性能提升没有任何帮助，反而浪费CPU时间，
2. 判断队列是否有压力的最简单办法就是打开RabbitMQ的管理界面，切换到 Queues 页， 查看队列的 Message/Total 列，如果一段时间内（1分钟），这个数量没有上升趋势， 就表示消息订阅者的数量是足够的。
3. 订阅者数量较小也不行，会导致消息的大量堆积，也会让MQ越来越慢。

所以，订阅者的数量以 “消息不堆积” 为原则。
这也是为什么订阅者的数量要做成配置参数的原因。

“消息不堆积” 举例说明

这张图片很明显 incoming 远大于deliver，因此 Total 越来越大，也就是消息堆积越来越多。

在这张图片中，incoming和deliver基本上相当，结果就是 Total 持续保持在较低水平，表示消息没有堆积。

指标监控

使用消息队列的应用程序，至少应该实现对队列的监控统计，主要有3个指标：

1. 消息积压数量
2. 当天消息处理总量
3. 消息吞吐量

例如：

实现过程可参考： 业务指标监控开发

使用 Kafka

本文将介绍Kafka的使用技巧，主要包含以下内容

1. 在配置服务中注册连接
2. 发送消息
3. 订阅消息
4. 注意事项

在配置服务中注册连接

在使用 Kafka 前，请先在配置服务中注册连接参数，可参考下图：

发送消息

示例代码：

[HttpPost]
[Route("send1.aspx")]
public int Send1(Product product)
{
    int index = Interlocked.Increment(ref s_index);
    product.Remark = $"Kafka message - {index}, {DateTime.Now.ToTimeString()}";

    using( KafkaClient client = new KafkaClient("Nebula.Log.Kafka") ) {
        return client.SendMessageNoWait(product);
    }
}

订阅消息

KafkaSubscriber.Start<Product, PrintMessageHandler>(new KafkaSubscriberArgs {
    SettingName = "Nebula.Log.Kafka",   // 连接的配置名称
    GroupId = null, // null 表示使用当前应用的名称 
    Topic = null,   // null 表示使用Product的类型全名代替
    CommitPeriod = 1,  // 自动提交周期，1 表示每处理1条消息就提交一次
    RetryCount = 0,    // 重试次数
    RetryWaitMilliseconds = 1000   // 重试的间隔时间
});

订阅消息之后，如果收到消息会交给 PrintMessageHandler 来处理，

后面的过程请参考：消息管道

注意事项

Kafka在整个消息处理过程中很有可能会导致消息丢失：

1. KafkaClient.SendMessageNoWait 方法，它只是将消息放入客户端的缓冲区队列，并不是立即发送， 而是由另外的后台线程执行发送，这样做的好处是： 异步可提高吞吐量，但是在应用程序重启/停止的时候，极有可能出现消息丢失。
2. 服务端在收到消息后，由于没有严格的事务保证，也是有可能导致消息丢失的
3. 消息在服务端只是写入日志文件，并不关心客户端的处理进度（不是常规的队列概念），在消息达到清理时间后，即使消息没有被处理，也会被清除

对于订阅者（消费端）来说，请注意：

1. KafkaSubscriber默认采用自动提交方式，可以设置 EnableAutoCommit = false 来使用【批量提交】
2. 通常为了提高所谓的吞吐量，会采用【批量提交】方式，例如：CommitPeriod = 10 ， 如果前面8条消息处理成功，处理第9条时程序重启， 当重新启动订阅后，前面的8条消息会重新发给订阅者，因为必须要实现【幂等操作】

参考链接

• Kafka Client Github： https://github.com/confluentinc/confluent-kafka-dotnet

• 官方文档，发送消息： https://developer.confluent.io/get-started/dotnet/#build-producer

• 官方文档，订阅消息： https://developer.confluent.io/get-started/dotnet/#build-consumer

使用 Pulsar

本文将介绍Pulsar的使用技巧，主要包含以下内容

1. 在配置服务中注册连接
2. 发送消息
3. 订阅消息

在配置服务中注册连接

在使用 Pulsar 前，请先在配置服务中注册连接参数，可参考下图：

4个参数项的含意：

• ServiceUrl: 用发送和订阅消息，具体格式要求请参考DotPulsar的要求
• WebServiceUrl: REST的访问地址，如果不指定Venus将不监控它的可用性
• AuthToken：认证用的 JWT Token
• RetryIntervalMs: The time to wait before retrying an operation or a reconnect. 可以不指定！

发送消息

示例代码：

private async Task<string> Send0(Product product, string topic)
{
    RuntimeData.PulsarCount.Increment();

    int index = Interlocked.Increment(ref s_index);
    product.Remark = $"Pulsar message - {index}, {DateTime.Now.ToTimeString()}";

    await using( NPulsarClient client = new NPulsarClient("Pulsar_test", topic) ) {
        var messageId = await client.SendMessageAsync(product);
        return messageId.ToString();
    }
}

订阅消息

PulsarSubscriber.StartAsync<Product, PrintMessageHandler2>(new PulsarSubscriberArgs {
    SettingName = "Pulsar_test",   // 连接的配置名称
    Topic = "product",
    SubscriberCount = 2,  // 订阅者数量
    RetryCount = 5,    // 重试次数
    RetryWaitMilliseconds = 1000   // 重试的间隔时间
});

订阅消息之后，如果收到消息会交给 PrintMessageHandler 来处理，

后面的过程请参考：消息管道

压缩解压缩

相关解释

• Pulsar服务端不处理压缩与解压缩
• 压缩由发送端实现，可选择压缩算法
• 解压缩由消费端在读取消息时判断，并自动调用对应的解压缩算法

Pulsar客户端支持4种压缩算法，可参考DotPulsar.CompressionType枚举，
相应的算法依赖包采用反射的方式加载，在运行时如果找不到合适的算法包会出现异常，
所以建议在程序部署时先添加必要的引用。

NPulsarClient默认使用Lz4压缩算法，如果需要使用其它算法，可在构造时指定：

/// <summary>
/// 构造方法
/// </summary>
/// <param name="settingName">配置服务中的Kafka连接名称</param>
/// <param name="topic">消息主题</param>
/// <param name="compressionType">压缩算法</param>
public NPulsarClient(string settingName, string topic, CompressionType compressionType = CompressionType.Lz4)

参考链接

• Pulsar C# client: https://pulsar.apache.org/docs/zh-CN/client-libraries-dotnet/

• Client Compression: https://github.com/apache/pulsar-dotpulsar/wiki/Compression

使用 MMQ

MMQ 即 Memory Message Queue，它是一种进程内的消息队列，由ClownFish提供。

它有以下特点：

• 性能好
• 不做持久化

创建MMQ实例

示例代码如下：

/// <summary>
/// 同步版本的 MMQ
/// </summary>
private static readonly MemoryMesssageQueue<Product> s_syncMMQ 
                            = new MemoryMesssageQueue<Product>(MmqWorkMode.Sync);

/// <summary>
/// 异步版本的 MMQ
/// </summary>
private static readonly MemoryMesssageQueue<Product> s_asyncMMQ 
                            = new MemoryMesssageQueue<Product>(MmqWorkMode.Async);

发送消息

示例代码-同步版本：

[HttpPost]
[Route("send-sync.aspx")]
public int Send1(Product product)
{
    int index = Interlocked.Increment(ref s_index);
    product.Remark = $"MMQ message - {index}, {DateTime.Now.ToTimeString()}";

    s_syncMMQ.Write(product);   // 注意这行

    return index;
}

示例代码-异步版本：

[HttpPost]
[Route("send-async.aspx")]
public async Task<int> Send2(Product product)
{
    int index = Interlocked.Increment(ref s_index);
    product.Remark = $"MMQ message - {index}, {DateTime.Now.ToTimeString()}";

    await s_asyncMMQ.WriteAsync(product);   // 注意这行

    return index;
}

订阅消息

// 同步版本
MmqSubscriber.Start<Product, PrintMessageHandler>(new MmqSubscriberArgs<Product> {
    Queue = s_syncMMQ,
    SubscriberCount = 2,
    RetryCount = 0,    // 重试次数
    RetryWaitMilliseconds = 1000   // 重试的间隔时间
});

// 异步版本
MmqSubscriber.StartAsync<Product, PrintMessageHandler2>(new MmqSubscriberArgs<Product> {
    Queue = s_asyncMMQ,
    SubscriberCount = 2,
    RetryCount = 0,    // 重试次数
    RetryWaitMilliseconds = 1000   // 重试的间隔时间
});

订阅消息之后，如果收到消息会交给 PrintMessageHandler 来处理，

后面的过程请参考：消息管道

使用 Redis MQ

ClownFish对Redis的封装并不多，主要在4个方面：

• 连接的统一管理
• API调用性能监控
• 发布消息
• 订阅消息

在配置服务中注册连接

在使用Redis前，请先在在配置服务中注册连接，可参考以下界面：

此处的参数值格式可参考：https://stackexchange.github.io/StackExchange.Redis/Configuration#basic-configuration-strings

调用 Redis API

可参考以下代码

[HttpGet, HttpPost]
[Route("string-demo.aspx")]
public async Task<object> StringDemo()
{
    IDatabase db = ClownFish.NRedis.Redis.GetDatabase();

    await db.StringSetAsync("kk101", DateTime.Now.Ticks.ToString());

    await db.StringSetAsync("kk102", Guid.NewGuid().ToString());

    string value1 = await db.StringGetAsync("kk101");

    string value2 = await db.StringGetAsync("kk102");

    return new {
        kk101 = value1,
        kk102 = value2
    };
}

核心部分就是得到 IDatabase实例

IDatabase db = ClownFish.NRedis.Redis.GetDatabase();

剩下的就是调用 Redis API

发布消息

示例代码：

[HttpPost]
[Route("sendmessage.aspx")]
public int Send(Product product)
{
    int index = Interlocked.Increment(ref s_index);
    product.Remark = $"Redis message - {index}, {DateTime.Now.ToTimeString()}";

    // channel 取Product的类型全名
    Redis.GetClient().SendMessage(product);

    return index;
}

订阅消息

RedisSubscriber.Start<Product, PrintMessageHandler>(new RedisSubscriberArgs {
    SettingName = "redis.Connection",   // 连接的配置名称
    Channel = null,   // null 表示使用Product的类型全名代替
    RetryCount = 5,    // 重试次数
    RetryWaitMilliseconds = 1000   // 重试的间隔时间
});

注意事项

• 不要把Redis当成数据库，它只是一个缓存服务！
• 所有写入Redis的数据都需要 设置过期时间
• 禁止将较大的数据块写入Redis，尽量控制数据长度在256K以内。

可靠地调用第三方

调用第三方服务可能看起来很简单：只需要发起一个HTTP请求即可！

实际上，如果真这样去做，迟早会出现各种稳定性或者数据不一致的问题，因为：

• 网络有可能偶尔中断，HTTP调用就会失败。
• 第三方可能在更新或者重启或者崩溃了，HTTP调用出现失败。

最麻烦的是：这种失败不是简单的 for循环+重试就能解决，因为不知道故障会持续多久！

调用第三方通常有2大场景：

• 即时调用（同步等待调用）：需要立即获取结果，然后继续执行
• 当前任务执行结束，【通知】第三方做相应处理

解决办法：

• 第一种场景：建议使用 for循环+重试，或者使用 ClownFish中的 HttpRetry 来解决。
• 第二种场景：有2种方法
  • 使用 WebHook服务 来代发通知
  • 使用 异步任务 + MQ 来实现

以上2种方法都依赖于 MQ 和 消息管道，差别：

• 前者：开发简单，但是需要多部署一个服务以及相关配置
• 后者：灵活度高

请点击目录中的节点查看具体文档

后台任务介绍

后台任务分为2大类：

• 周期性后台任务
• 临时性后台任务

本文主要介绍 周期性后台任务

周期性后台任务

很多时候我们需要执行一些周期性任务，它们有如下特性：

• 后台执行：由后台线程执行
• 周期规律性：每x分钟 或者 每x秒 就执行一次
• 长时间运行：任务的生命周期和进程相同

对于此类场景，我们可以使用 BackgroundTask 来实现。

周期性后台任务有2种执行方式：

• 同步方式：需要从 BackgroundTask 基类继承
• 异步方式：需要从 AsyncBackgroundTask 基类继承

2个类的用法完全一致，下文将以同步方法来介绍用法。

BackgroundTask与Hangfire的主要差别

查看后台任务

可以从Venus中查看某个应用进程的后台作业执行情况，可参考下图：

周期性后台任务-同步版本

所有周期性后台任务需要实现 以下任意一个 抽象类：

• BackgroundTask
• AsyncBackgroundTask

BackgroundTask

BackgroundTask 是一个基类，当我们在项目中实现它的子类时，就能实现【后台周期性任务】，
AsyncBackgroundTask 是对应的异步版本。

它的主要公开成员如下：

/// <summary>
/// 表示一个后台运行的任务，它的子类会在程序启动时自动创建并运行
/// </summary>
public abstract class BackgroundTask
{
	/// <summary>
	/// 是否需要在每次执行Execute方法时自动生成日志(OprLog + InvokeLog)
	/// </summary>
	protected bool EnableLog => true;

	/// <summary>
	/// true表示需要在启动任务线程后立即执行一次，false表示启动后等到到达执行周期才会执行。
	/// 默认值：false
	/// </summary>
	public virtual bool FirstRun => false;
	
	/// <summary>
	/// 获取休眠秒数，用于描述“周期任务”的间隔时间，例如：每5秒执行一次
	/// 注意：同步版本不支持时间跨度太久的休眠间隔。
	/// </summary>
	public virtual int? SleepSeconds {
		get => null;
	}

	/// <summary>
	/// 获取一个Cron表达式，用于描述“周期任务”的间隔时间。
	/// 这里使用的是 Quartz 支持的 Cron 格式，在线工具：https://www.pppet.net/
	/// 注意：同步版本不支持时间跨度太久的休眠间隔。
	/// </summary>
	public virtual string CronValue {
		get => null;
	}

	/// <summary>
	/// 执行任务前的初始化。
	/// 说明：执行当前方法时框架不做异常处理，如果产生异常会导致进程崩溃。
	/// </summary>
	/// <returns>如果 return false, 表示初始化失败，将中止任务</returns>
	public virtual bool Init()


	/// <summary>
	/// 执行任务的主体过程。
	/// </summary>
	public abstract void Execute();

	/// <summary>
	/// 异常处理方法。
	/// 默认行为：如果启用日志就不做任何处理，否则输出到Console
	/// </summary>
	/// <param name="ex"></param>
	public virtual void OnError(Exception ex)
	
}

要求&说明

子类的实现要求：

• 类型的可见性必须是 public
• 必须指定执行间隔属性： SleepSeconds 或者 CronValue

补充说明：

• 子类的实例由Nebula在启动时创建并调用（项目中不需要实例化）
• 非Nebula的项目可以调用 BackgroundTaskManager.StartAll 方法来启动任务
• 内部已包含异常处理，如果需要额外的日志请重写OnError方法

BackgroundTask, AsyncBackgroundTask 差异

• AsyncBackgroundTask 是异步版本，使用线程池执行作业
• BackgroundTask 使用单独线程（不使用线程池），任务的及时触发率会更好
• 如果对任务的【及时触发】要求不高，建议使用 AsyncBackgroundTask

• AsyncBackgroundTask 支持较长的作业执行间隔，例如：一个月一次
• BackgroundTask 【不支持】较长的作业执行间隔，建议 仅用于 10 秒内的间隔任务

示例1 - 休眠N秒的定时任务

public class Task1 : BackgroundTask
{
    public override int? SleepSeconds => 90;

    public override void Execute()
    {
        Console2.Info("Task1，每隔 90 秒执行一次！");
		// do something....
    }
}

示例2 - 基于CronValue的定时任务

public class Task2 : BackgroundTask
{
    public override string CronValue => "0/10 * * * * ? *";

    public override void Execute()
    {
        Console2.Info("Task2，每隔 10 秒执行一次！");
		// do something....
    }
}

高级用法

1，线上调整作业执行周期

上面2个示例中，我们使用了【固定】的执行周期，

public override int? SleepSeconds => 90;
public override string CronValue => "0/10 * * * * ? *";

这种做法只是简单，不够灵活：它【不支持】线上调整作业执行周期

推荐做法：

private static readonly int s_sleepSeconds = LocalSettings.GetInt("XXXXXXXXXX_SleepSeconds", 90);

public override int? SleepSeconds => s_sleepSeconds;

• 以后线上需要调整作业的执行周期时，只需要添加一个环境变量就可以了。
• CronValue 也可以采用类似的思路，但是建议用Base64编码

SleepSeconds 的实现还可以更复杂，实现动态的作业执行周期：

• 白天(8:00 -- 20:00 ) 时间段内，作业的执行周期是　60 秒
• 夜间(白天之外的时间)时间段内，作业的执行周期是　120 秒

2，作业在程序启动后立即执行，但是延迟一小段时间

public class Task3 : AsyncBackgroundTask
{
    public override int? SleepSeconds => 90;

	public override bool FirstRun => true;   // 作业在程序启动后立即执行

    public override async Task ExecuteAsync()
    {
        if( this.ExecuteCount == 1 ) {   // 第一次运行时，延迟一段时间
            await Task.Delay( (new Random()).Next(10, 30) * 1000 );
        }

        Console2.Info("Task1，每隔 90 秒执行一次！");
		// do something....
    }
}

这里使用了2个基类的属性

• FirstRun ： 可参考前面的注释描述
• ExecuteCount ：它指示当前 Execute/ExecuteAsync 是第几次运行

周期性后台任务-异步版本

注意事项：

• 基类是 AsyncBackgroundTask
• 方法的签名是 Task ExecuteAsync()

示例1 - 休眠N秒的定时任务

public class Task1 : AsyncBackgroundTask
{
    public override int? SleepSeconds => 90;

    public override Task ExecuteAsync()
    {
        Console2.Info("Task1，每隔 90 秒执行一次！");
		// do something....

		await xxxxAsync();
    }
}

示例2 - 基于CronValue的定时任务

public class Task2 : AsyncBackgroundTask
{
    public override string CronValue => "0/10 * * * * ? *";

    public override Task ExecuteAsync()
    {
        Console2.Info("Task2，每隔 10 秒执行一次！");
		// do something....

		await xxxxAsync();
    }
}

临时性后台任务

针对一些复杂耗时的用户操作场景，一个常见的优化性能手段就是

• 先检查必要的提交数据是否完整有效（同步执行）
• 将复杂耗时部分改为异步执行（提前结束）

那么这里的【异步执行】就有多种实现方式了，例如：

• 使用 Task.Run(...)
• 使用 Hangfire 之类的组件
• 将所有数据参数打包成一个消息结构，发送到MQ，再订阅处理

前2种方式都是 不建议 的做法，因为：

• Task.Run，它其实是将操作任务提交到线程池，这里就有4个延伸问题：
  • 与HTTP请求争用线程，影响进程的吞吐量，甚至能出现HTTP503
  • 程序可能随时重启或者崩溃，没办法保证异步任务能执行结束
  • 没有日志
  • 缺乏监控
• Hangfire，在这种场景下（临时任务）它其实是在解决Task.Run的问题，但是它有2大缺点
  • 增加数据库压力，尤其是非SQLSERVER数据库
  • 日志太弱，参考价值较小。

推荐做法

推荐做法：2个步骤

1. 将所有数据参数打包成一个消息结构，发送到MQ
2. 订阅MQ，根据消息内容执行具体任务

为什么是这样：

• 消息一旦进入MQ，就不会丢失，程序重启或者崩溃都不受影响
• 不依赖于数据库或者Redis，它们容易出现性能瓶颈问题
• 充分利用ClownFish消息管道特性：
  • 重试
  • 多订阅者（数量可调，压力恒定）
  • 日志记录
  • 性能监控

多租户后台作业建议做法

作业特点

这类任务有2个明显特点：

• 对所有租户执行相同的操作，例如：执行SQL脚本更新一些数据表
• 执行时间长（因为第1点）

不建议的实现方式

以下用伪码举例

// 获取所有租户ID清单
List<string> list = GetAllTenantId();

// 用StringBuilder来存储所有租房的处理过程
StringBuilder sb = new ();

foreach(string tenantId in list) {

    // 在循环中一次处理一个租户
    string text = ProcessTenant(tenantId);
    sb.AppendLine(text);
}

// 给外界发送通知
SendNotify(sb.ToString());

这样设计有以下缺点：

• 不利于任务重试！
  • 重试成本过高，重试只能从头开始
• 不能中断！
• 不利于性能评估！
  • 因为循环所有租户耗费的时间太长，不知道在某个租户上的具体耗时。
  • 带来一个新问题：由于没有性能监控，也 没法做性能优化！

推荐的实现方式

核心原则：能方便地评估代码的执行性能！

实现过程：

• 拆分成2个部分
• 主任务只做分裂处理（不做业务处理）
  • 如果是周期性的任务，则用 BackgroudTask 来实现
  • 如果是临时性的用户触发导致，可以直接直接由 HttpAction 实现

  ---

  • 每次执行时，按租户拆分成多个子任务
  • 每个子任务由一条消息表示，其中包含一个租户ID，然后将消息发送到MQ
• 子任务（一定是后台运行）
  • 用 MessageHandler 来实现（有日志和性能监控）
  • 根据消息中的租户ID，完成对一个租户的业务处理（任务足够小，性能评估才有意义）

以下是实现步骤的伪码。

1，先定义消息类型

public class TenantMssage {
    public string TenantId {get; set;}
    public string 其它所需的参数 {get; set;}


    // 用于记录执行过程中产生的消息（此属性不是必需）
    public StringBuilder Messages;

    // 用于记录执行了多少次（此属性不是必需）
    public ValueCounter ExecCounter;
}

2，程序初始化时注册消息订阅者

private static readonly MemoryMesssageQueue<TenantMssage> s_asyncMMQ 
                            = new MemoryMesssageQueue<TenantMssage>(MmqWorkMode.Async);

MmqSubscriber.StartAsync<TenantMssage, TenantMssageHandler>(new MmqSubscriberArgs {
    Queue = s_asyncMMQ,
    SubscriberCount = LocalSettings.GetUInt("TenantMssageHandler_Subscriber_Count", 1)
});

3-1，临时性主任务的实现-HttpAction

[Route("test1.aspx")]
public int Test1()
{
    // 获取要处理的租户ID清单
    List<string> list = GetAllTenantId();

    foreach(string tenantId in list) {
            TenantMssage msg = new TenantMssage{ TenantId = tenantId };
            client.SendMessage(msg);        
        }
}

3-2，周期性主任务的实现-BackgroudTask

public class TenantTask1 : BackgroundTask
{
    public override int? SleepSeconds => 3600;    // 假设1小时执行一次

    public override void Execute()
    {
        // 获取要处理的租户ID清单
        List<string> list = GetAllTenantId();

        using( RabbitClient client = new RabbitClient("SettingName", "Application111") ) {
            foreach(string tenantId in list) {
                TenantMssage msg = new TenantMssage{ TenantId = tenantId };
                client.SendMessage(msg);        
            }
        }
    }
}

4，执行租户处理逻辑-MessageHandler

public class TenantMssageHandler : BaseMessageHandler<TenantMssage>
{
    public override void ProcessMessage(PipelineContext<TenantMssage> context)
    {
        // 获取消息对象
        TenantMssage message = context.MessageData;
        
        ProcessTenant(message);
    }
}

任务的整体执行进度

如果需要获取任何的整体执行进度，可以参考下面的实现方式。

1，在消息结构中，增加一个总数，例如：

public class TenantMssage {
    public string TenantId {get; set;}
    public int TenantCount {get; set;}
    public string 其它所需的参数 {get; set;}
}

2，开始执行任务时

[Route("test1.aspx")]
public int Test1()
{
    // 获取要处理的租户ID清单
    List<string> list = GetAllTenantId();

    // ========================================= 注意下面这行代码
    // 创建一个计数器，用于累加执行进度
    Redis.GetDatabase().StringSet("XXX业务场景_key", 0);
    // =========================================

    using( RabbitClient client = new RabbitClient("SettingName", "Application111") ) {
        foreach(string tenantId in list) {
            TenantMssage msg = new TenantMssage{ TenantId = tenantId };
            client.SendMessage(msg);        
        }
    }
}

3，在消息处理时，用Redis的计数器功能累加进度：

public class TenantMssageHandler : BaseMessageHandler<TenantMssage>
{
    public override void ProcessMessage(PipelineContext<TenantMssage> context)
    {
        // 获取消息对象
        TenantMssage message = context.MessageData;
        
        ProcessTenant(message);

        // 更新执行进度
        Redis.GetDatabase().StringIncrement("XXX业务场景_key");
    }
}

4，获取整体执行进度

int count = Redis.GetDatabase().StringGet("XXX业务场景_key");

int 百分比 = count / TenantCount;

// 如果到了 100%，可以把这个 Redis 计数器删除。

工具类

ClownFish提供了很多工具类，用于一些特定场景。
由于功能比较杂，本文仅列出 部分 工具类的名称，实际使用时可参考VS的智能提示。

• CacheItem/CacheDictionary: 用于缓存操作
• AliWebApiQueryBuilder: 用于构造阿里API-URL，不必引用他们的SDK
• EnvUtils：获取运行环境信息
• GuidHelper： 可生成有序GUID
• MyTimeZone：时区处理
• StringConverter: 扩展类，提供一些字符串相关的转换功能
• TSafeDictionary：线程安全的字典集合
• UnixTime：UNIX时间相关的工具类
• ValueCounter： 计数器
• ByteExtensions：Byte 相关扩展工具类
• DataTableExtensions：DataTable 相关扩展工具类
• DateTimeExtensions：DateTime 相关扩展工具类
• DictionaryExtensions：Dictionary 相关扩展工具类
• FormatUtils：数字格式化
• IntExtensions：int 相关扩展工具类
• MethodExtensions：MethodInfo 相关扩展工具类
• StringExtensions：string 相关扩展工具类
• ReliableFile：实现文件的可靠读取工具类
• TempFile: 封装一个临时文件对象，在使用完后会自动清除
• ThreadUtils: 线程相关工具类
• UrlExtensions: URL相关工具类

序列化

ClownFish为XML和JSON序列化提供了便捷的扩展方法

XML序列化

/// <summary>
/// XML序列化与反序列化的扩展方法类
/// </summary>
public static class XmlExtensions
{
	/// <summary>
	/// 将对象执行XML序列化（使用UTF-8编码）
	/// </summary>
	/// <param name="obj">要序列化的对象</param>
	/// <returns>XML序列化的结果</returns>
	public static string ToXml(this object obj)


	/// <summary>
	/// 从XML字符串中反序列化对象（使用UTF-8编码）
	/// </summary>
	/// <typeparam name="T">反序列化的结果类型</typeparam>
	/// <param name="xml">XML字符串</param>
	/// <returns>反序列化的结果</returns>
	public static T FromXml<T>(this string xml)


	/// <summary>
	///  从XML字符串中反序列化对象（使用UTF-8编码）
	/// </summary>
	/// <param name="s"></param>
	/// <param name="type">反序列化的结果类型</param>
	/// <returns></returns>
	public static object FromXml(this string s, Type type)

}

JSON序列化

/// <summary>
/// JSON序列化的工具类
/// </summary>
public static class JsonExtensions
{
	/// <summary>
	/// 将一个对象序列化为JSON字符串。
	/// </summary>
	/// <param name="obj">要序列化的对象</param>
	/// <returns>序列化得到的JSON字符串</returns>
	public static string ToJson(this object obj)


	/// <summary>
	/// 将一个对象序列化为JSON字符串。
	/// </summary>
	/// <param name="obj">要序列化的对象</param>
	/// <param name="style">JSON序列化格式</param>
	/// <returns>序列化得到的JSON字符串</returns>
	public static string ToJson(this object obj, JsonStyle style)


	/// <summary>
	/// 将一个JSON字符串反序列化为对象
	/// </summary>
	/// <typeparam name="T">反序列的对象类型参数</typeparam>
	/// <param name="json">JSON字符串</param>
	/// <returns>反序列化得到的结果</returns>
	public static T FromJson<T>(this string json)


	/// <summary>
	/// 将一个JSON字符串反序列化为对象
	/// </summary>
	/// <param name="json">JSON字符串</param>
	/// <param name="destType">反序列的对象类型参数</param>
	/// <returns>反序列化得到的结果</returns>
	public static object FromJson(this string json, Type destType)
}

示例代码

public void Json_序列化_反序列化()
{
	var obj = new NameValue { Name = "aaaa", Value = "123" };

	// 将对象序列化成JSON字符串
	string json = obj.ToJson();

	// 将JSON字符串反序列化为指定的类型
	var object2 = json.FromJson<NameValue>();
}


public void Xml_序列化_反序列化()
{
	var obj = new NameValue { Name = "aaaa", Value = "123" };

	// 将对象序列化成XML字符串
	string xml = obj.ToXml();

	// 将XML字符串反序列化为指定的类型
	var object2 = xml.FromXml<NameValue>();
}


/// <summary>
/// 模拟一个配置类型
/// </summary>
public class RunOption { /* 这里省略具体的参数成员 */ }

public void Xml_配置文件_序列化_反序列化()
{
	RunOption option = new RunOption();
	string file = @"d:\aa\xx.config";

	// 将对象序列化成XML配置文件
	XmlHelper.XmlSerializeToFile(option, file);

	// 将XML配置文件反序列化为指定的类型
	option = XmlHelper.XmlDeserializeFromFile<RunOption>(file);
}

数据压缩

ClownFish为数据压缩提供了便捷的扩展方法

/// <summary>
/// GZIP压缩相关的工具方法
/// </summary>
public static class GzipHelper
{
	/// <summary>
	/// 用GZIP压缩一个字符串，并以BASE64字符串的形式返回压缩后的结果
	/// </summary>
	/// <param name="input"></param>
	/// <returns></returns>
	public static string Compress(string input)


	/// <summary>
	/// 用GZIP解压缩一个BASE64字符串
	/// </summary>
	/// <param name="base64"></param>
	/// <returns></returns>
	public static string Decompress(string base64)


	/// <summary>
	/// 将一个文本转指定编码后做Gzip压缩
	/// </summary>
	/// <param name="text"></param>
	/// <param name="encoding">默认UTF8</param>
	/// <returns></returns>
	public static byte[] ToGzip(this string text, Encoding encoding = null)


	/// <summary>
	/// 用Gzip压缩格式压缩一个二进制数组
	/// </summary>
	/// <param name="input"></param>
	/// <returns></returns>
	public static byte[] ToGzip(this byte[] input)


	/// <summary>
	/// 解压缩一个二进制数组
	/// </summary>
	/// <param name="input"></param>
	/// <returns></returns>
	public static byte[] UnGzip(this byte[] input)
}

ZIP文件压缩

示例代码

public void 将ZIP文件释放到指定目录()
{
	string zipPath = @"D:\my-github\bin.zip";
	string destPath = @"D:\test";

	ZipHelper.ExtractFiles(zipPath, destPath);
}

public void 在内存中读取ZIP文件()
{
	string zipPath = @"D:\my-github\bin.zip";

	var data = ZipHelper.Read(zipPath);

	// 结果是一个元组LIST
	// 每个元组中，Item1 就是压缩包内的文件名，Item2 就是文件内容。
	// 如果需要读取某个文件，可以从LIST中查找
}

public void 将指定目录打包成ZIP文件_含子目录()
{
	string srcPath = @"D:\test";
	string zipPath = @"D:\my-github\bin.zip";

	ZipHelper.CompressDirectory(srcPath, zipPath);
}

public void 将内存数据打包成ZIP文件()
{
	List<ZipItem> data = null;    // 需要准备好这个结构
	string zipPath = @"D:\my-github\bin.zip";

	ZipHelper.Compress(data, zipPath);
}

重试

对于一些可能会出现失败的操作，我们可以采用重试策略来执行，

例如：

• 读写文件
• 发送HTTP请求

示例代码

示例代码

public void 发送HTTP请求_支持重试()
{
    // 如果发生异常，最多重试5次，间隔 500 毫秒
    Retry retry = Retry.Create(5, 500);

    string text = new HttpOption {
        Url = "http://www.fish-test.com/abc.aspx",
        Data = new { id = 2, name = "abc" }
    }.GetResult(retry);
}

public void 发送HTTP请求_定制重试策略()
{
    Retry retry =
        Retry.Create(100, 2000)     // 重试 100 次，间隔 2 秒
        .Filter<WebException>();    // 仅当出现 WebException 异常时才重试。

    string text = new HttpOption {
        Url = "http://www.fish-test.com/abc.aspx",
        Data = new { id = 2, name = "abc" }
    }.GetResult(retry);
}

public void 读取文件_最大重试10次()
{
    string text =
        Retry.Create(100, 0) // 指定重试10，使用默认的重试间隔时间，且不分辨异常类型（有异常就重试）
            .Run(() => {
                return System.IO.File.ReadAllText(@"c:\abc.txt", Encoding.UTF8);
            });
}

工具类

ClownFish提供4个封装的工具可直接使用。

    /// <summary>
    /// 提供一些与 System.IO.File 相同签名且功能相同的工具方法，
    /// 差别在于：当出现IOException时，这个类中的方法支持重试功能。
    /// </summary>
    public static class RetryFile

    /// <summary>
    /// 提供一些与 System.IO.Directory 相同签名且功能相同的工具方法，
    /// 差别在于：当出现IOException时，这个类中的方法支持重试功能。
    /// </summary>
    public static class RetryDirectory

/// <summary>
/// 定义HttpClient的扩展方法的工具类
/// </summary>
public static class HttpOptionExtensions
{
	/// <summary>
	/// 根据指定的HttpOption参数，用【同步】方式发起一次HTTP请求
	/// </summary>
	/// <param name="option">HttpOption的实例，用于描述请求参数</param>
	/// <param name="retry">提供一个Retry实例，用于指示如何执行重试。如果此参数为NULL则不启用重试</param>
	/// <returns>返回服务端的调用结果，并转换成指定的类型</returns>
	/// <exception cref="RemoteWebException"></exception>
	public static string GetResult(this HttpOption option, Retry retry = null)


	/// <summary>
	/// 根据指定的HttpOption参数，用【异步】方式发起一次HTTP请求
	/// </summary>
	/// <param name="option">HttpOption的实例，用于描述请求参数</param>
	/// <param name="retry">提供一个Retry实例，用于指示如何执行重试。如果此参数为NULL则不启用重试</param>
	/// <returns>返回服务端的调用结果，并转换成指定的类型</returns>
	/// <exception cref="RemoteWebException"></exception>
	public async static Task<string> GetResultAsync(this HttpOption option, Retry retry = null)


	/// <summary>
	/// 根据指定的HttpOption参数，用【同步】方式发起一次HTTP请求
	/// </summary>
	/// <typeparam name="T">返回值的类型参数</typeparam>
	/// <param name="option">HttpOption的实例，用于描述请求参数</param>
	/// <param name="retry">提供一个Retry实例，用于指示如何执行重试。如果此参数为NULL则不启用重试</param>
	/// <returns>返回服务端的调用结果，并转换成指定的类型</returns>
	/// <exception cref="RemoteWebException"></exception>
	public static T GetResult<T>(this HttpOption option, Retry retry = null)


	/// <summary>
	/// 根据指定的HttpOption参数，用【异步】方式发起一次HTTP请求
	/// </summary>
	/// <typeparam name="T">返回值的类型参数</typeparam>
	/// <param name="option">HttpOption的实例，用于描述请求参数</param>
	/// <param name="retry">提供一个Retry实例，用于指示如何执行重试。如果此参数为NULL则不启用重试</param>
	/// <returns>返回服务端的调用结果，并转换成指定的类型</returns>
	/// <exception cref="RemoteWebException"></exception>
	public async static Task<T> GetResultAsync<T>(this HttpOption option, Retry retry = null)

}

/// <summary>
/// 创建发出HTTP请求时的重试策略工具类
/// </summary>
public static class HttpRetry
{
	/// <summary>
	/// 获取用于发送HTTP请求的重试策略。
	/// 当发生 “网络不通” 或者 “HTTP 502,503” 时，最大重试 7 次，每次间隔 1000 毫秒
	/// </summary>
	/// <returns></returns>
	public static Retry Create()
}

常识科普

• 加密 和 哈希 不是一回事 ！！！
• 加密后的结果一定是可以解密的，也就是说可以根据结果得到加密前的输入内容
• 哈希又称为单向散列，是一种计算消息摘要的方法，过程不可逆，不能根据结果得到哈希前的输入
• 常见的哈希算法有：md5, sha1, sha256, sha512
• 加密算法分为：对称加密算法 和 非对称加密算法
• 常见的对称加密算法有：3DES, AES
• 常见的非对称加密算法有：RSA
• RSA算法的用途有：加密，解密，计算签名，验证签名
• RSA算法的密钥分为：公钥（用于加密/计算签名）， 私钥（用于解密/验证签名）
• RSA算法的 公钥通常用 cer 文件存储，或者直接在代码中硬编码多行的BASE64密钥文本
• RSA算法的 私钥通常用 pfx 文件存储，此文件需要密码才能被加载到内存
• RSA的密钥可以导入操作系统的证书管理器，然后通过指纹引用，此时不需要密码（即使是私钥）
• RSA的私钥通常不随客户端程序公开发布，私钥一定要保留在服务端，否则可以考虑使用 对称加密算法
• RSA的加密算法只能加密小文本，通常是密码之类的敏感内容，长文本要结合对称加密算法做二次加密

Aes加密解密

/// <summary>
/// 对AES算法封装的工具类
/// </summary>
public static class AesHelper
{
	/// <summary>
	/// 使用AES算法加密字符串
	/// </summary>
	/// <param name="text"></param>
	/// <param name="password"></param>
	/// <returns></returns>
	public static string Encrypt(string text, string password)

	/// <summary>
	/// 使用AES算法加密字节数组
	/// </summary>
	/// <param name="input"></param>
	/// <param name="password"></param>
	/// <returns></returns>
	public static byte[] Encrypt(byte[] input, string password)

	/// <summary>
	/// 使用AES算法解密一个以Base64编码的加密字符串
	/// </summary>
	/// <param name="base64"></param>
	/// <param name="password"></param>
	/// <returns></returns>
	public static string Decrypt(string base64, string password)

	/// <summary>
	/// 使用AES算法解密字节数组
	/// </summary>
	/// <param name="input"></param>
	/// <param name="password"></param>
	/// <returns></returns>
	public static byte[] Decrypt(byte[] input, string password)
}

Hash

/// <summary>
/// 封装常用的HASH算法
/// </summary>
public static class HashHelper
{
	/// <summary>
	/// 计算字符串的 SHA1 签名
	/// </summary>
	/// <param name="text"></param>
	/// <param name="encoding"></param>
	/// <returns></returns>
	public static string Sha1(this string text, Encoding encoding = null)


	/// <summary>
	/// 计算字符串的 SHA256 签名
	/// </summary>
	/// <param name="text"></param>
	/// <param name="encoding"></param>
	/// <returns></returns>
	public static string Sha256(this string text, Encoding encoding = null)


	/// <summary>
	/// 计算字符串的 SHA512 签名
	/// </summary>
	/// <param name="text"></param>
	/// <param name="encoding"></param>
	/// <returns></returns>
	public static string Sha512(this string text, Encoding encoding = null)


	/// <summary>
	/// 计算文件的SHA1值
	/// </summary>
	/// <param name="filePath"></param>
	/// <returns></returns>
	public static string FileSha1(string filePath)


	/// <summary>
	/// 计算字符串的 MD5 签名
	/// </summary>
	/// <param name="text"></param>
	/// <param name="encoding"></param>
	/// <returns></returns>
	public static string Md5(this string text, Encoding encoding = null)


	/// <summary>
	/// 计算文件的MD5值
	/// </summary>
	/// <param name="filePath"></param>
	/// <returns></returns>
	public static string FileMD5(string filePath)
}

RSA 证书

/// <summary>
/// 包含一些查找X509证书的工具方法
/// </summary>
public static class X509Finder
{
	/// <summary>
	/// 根据证书指纹查找X509证书，优先查找LocalMachine存储区域，如果失败则再查找CurrentUser
	/// </summary>
	/// <param name="thumbprint">证书指纹</param>
	/// <param name="ifNotFoundThrowException">如果没有找到证书是否需要抛出异常</param>
	/// <returns></returns>
	public static X509Certificate2 FindByThumbprint(string thumbprint, bool ifNotFoundThrowException)



	/// <summary>
	/// 根据指定的证书指纹和位置，查找证书。
	/// </summary>
	/// <param name="thumbprint">证书指纹</param>
	/// <param name="storeLocation"></param>
	/// <param name="storeName"></param>
	/// <returns></returns>
	public static X509Certificate2 FindByThumbprint(string thumbprint,
													StoreLocation storeLocation = StoreLocation.LocalMachine,
													StoreName storeName = StoreName.My)


	/// <summary>
	/// 从一个公钥字符串中加载X509证书
	/// </summary>
	/// <param name="publicKey"></param>
	/// <returns></returns>
	public static X509Certificate2 LoadFromPublicKey(byte[] publicKey)


	/// <summary>
	/// 从一个公钥字符串中加载X509证书
	/// </summary>
	/// <param name="publicKeyText"></param>
	/// <returns></returns>
	public static X509Certificate2 LoadFromPublicKey(string publicKeyText)


	/// <summary>
	/// 从一个公钥文件中加载X509Certificate2
	/// </summary>
	/// <param name="publicKeyFilePath"></param>
	/// <returns></returns>
	public static X509Certificate2 LoadPublicKeyFile(string publicKeyFilePath)


	/// <summary>
	/// 从pfx文件内容中加载一个X509Certificate2对象
	/// </summary>
	/// <param name="pfxBody"></param>
	/// <param name="password"></param>
	/// <returns></returns>
	public static X509Certificate2 LoadPfx(byte[] pfxBody, string password)


	/// <summary>
	/// 从pfx文件内容中加载一个X509Certificate2对象
	/// </summary>
	/// <param name="pfxFilePath"></param>
	/// <param name="password"></param>
	/// <returns></returns>
	public static X509Certificate2 LoadPfx(string pfxFilePath, string password)
}

RSA算法

/// <summary>
/// RSA算法（签名/验证签名/加密/解密）的封装工具类
/// </summary>
public static class X509Extensions
{
	/// <summary>
	/// 用X509证书对数据做签名
	/// </summary>
	/// <param name="cert"></param>
	/// <param name="data"></param>
	/// <returns></returns>
	public static string Sign(this X509Certificate2 cert, byte[] data)


	/// <summary>
	/// 用X509证书验证数据签名
	/// </summary>
	/// <param name="cert"></param>
	/// <param name="data"></param>
	/// <param name="signature"></param>
	/// <returns></returns>
	public static bool Verify(this X509Certificate2 cert, byte[] data, string signature)


	/// <summary>
	/// 用X509证书加密数据。
	/// 注意：这个方法只能加密比较短的内容（一般是密钥）
	/// </summary>
	/// <param name="cert"></param>
	/// <param name="data"></param>
	/// <returns></returns>
	public static byte[] Encrypt(this X509Certificate2 cert, byte[] data)



	/// <summary>
	/// 用X509证书解密数据
	/// </summary>
	/// <param name="cert"></param>
	/// <param name="data"></param>
	/// <returns></returns>
	public static byte[] Decrypt(this X509Certificate2 cert, byte[] data)
}

高性能的反射工具类

ClownFish提供了以下工具来优化反射调用的性能：

对象构造

public static class CtorExtensions
{
	/// <summary>
	/// 快速实例化一个对象
	/// </summary>
	/// <param name="instanceType"></param>
	/// <returns></returns>
	public static object FastNew(this Type instanceType)

}

Field读写

public static class FieldInfoExtensions
{
	/// <summary>
	/// 快速读取字段值
	/// </summary>
	/// <param name="fieldInfo"></param>
	/// <param name="instance"></param>
	/// <returns></returns>
	public static object FastGetValue(this FieldInfo fieldInfo, object instance)


	/// <summary>
	/// 快速给字段赋值
	/// </summary>
	/// <param name="fieldInfo"></param>
	/// <param name="instance"></param>
	/// <param name="value"></param>
	public static void FastSetValue(this FieldInfo fieldInfo, object instance, object value)

}

Property读写

public static class PropertyInfoExtensions
{
	/// <summary>
	/// 快速读取属性值
	/// </summary>
	/// <param name="propertyInfo"></param>
	/// <param name="instance"></param>
	/// <returns></returns>
	public static object FastGetValue(this PropertyInfo propertyInfo, object instance)

	/// <summary>
	/// 快速给属性赋值
	/// </summary>
	/// <param name="propertyInfo"></param>
	/// <param name="instance"></param>
	/// <param name="value"></param>
	public static void FastSetValue(this PropertyInfo propertyInfo, object instance, object value)
}

方法调用

public static class MethodInfoExtensions
{
	/// <summary>
	/// 快速调用方法
	/// </summary>
	/// <param name="methodInfo"></param>
	/// <param name="instance"></param>
	/// <param name="parameters"></param>
	/// <returns></returns>
	public static object FastInvoke(this MethodInfo methodInfo, object instance, params object[] parameters)
}

日志表清理

虽然 Nebula.Juno 是一个通用的日志数据表清理服务，可以让你不写代码就能实现对日志表的清理，
但是有些场景下，你不希望部署这样一个服务，
那么可以使用 ClownFish 提供的API来实现同样的功能，可参考以下代码：

public class DataCleanWorker : BackgroundTask
{
    private static readonly int s_days = LocalSettings.GetInt("UserAskLogKeepDays", 10);

    public override int? SleepSeconds => 3600;

    public override bool FirstRun => true;

    public override void Execute()
    {
        CleaningOption option = new CleaningOption {
            DbConfig = MyConfig.DbConfig,
            TableName = "user_ask_log",
            TimeFieldName = "ask_time",
            HoursAgo = 24 * s_days  //  保留10天的日志
        };

        DataCleaner cleaner = new DataCleaner(option);

        try {
			// 执行日志清理动作
            cleaner.Execute();
        }
        finally {
            OprLog log = this.Context.OprLog;
            if( log != null ) {
                log.Text1 = cleaner.GetLogs();
            }
        }
    }
}

核心代码:

CleaningOption option = new CleaningOption { .... };
DataCleaner cleaner = new DataCleaner(option);
cleaner.Execute();

JWT

ClownFish内置了 JWT 的实现， 可用来生成或者解析 JWT-TOKEN

生成 JWT-TOKEN

var data = new {
	iss = "JwtUtilsTest",
	sub = "all",
	iat = DateTime.Now.ToNumber(),
	exp = DateTime.Now.AddDays(1).ToNumber(),
	UseId = 123,
	UserName = "Fish Li",
	UserRole = "Admin"
};

string json = data.ToJson();

string token = JwtUtils.Encode(json, JwtKey, "HS512");

或者

WebUserInfo user = new WebUserInfo {
    UserCode = "code111",
    UserName = "name123",
    UserId = "id_111",
    UserRole = "admin",
    TenantId = "tid_2222",
    TenantCode = "tcode_333",
    UserType = "type111"
};

int expirationSeconds = 3600 * 24;  // token 有效期 24 小时
string token = AuthenticationManager.GetLoginToken(user, expirationSeconds)

解析 JWT-TOKEN

绝大多数情况下，业务代码不需要 解析JWT-TOKEN，因为这个步骤属于 身份认证 的范畴，框架会自动处理。

有2类场景下可能会需要：

• MessageHandler
• BackgroundTask

此时可以参考以下示例：

string token = ...............; // 假设你已获取一个 jwt-token
string payload = JwtUtils.Decode(token, secretKey, algorithmName);

示例代码中的 payload 通常是一个 json 字符串，后面你只需要根据实现情况做反序列化即可。

说明：

• 如果参数 secretKey=null，那么将忽略签名验证
• Decode不做有效期检查，需要自行实现。原因：有效性的定义没有形成标准，ClownFish没法做！
• 框架自带的AuthenticateModule会始终执行 签名和有效期 验证。

支持的算法

JwtUtils工具类支持以下这些算法：

• HS256
• HS512
• RS256
• RS512
• ES256
• ES512

注意：后面4个是 非对称 算法，需要使用另外2个不同的方法

public static class JwtUtils
{
    public static string Encode2(string payload, X509Certificate2 x509, string algorithmName);

	public static string Decode2(string token, X509Certificate2 x509, string algorithmName);
}

注意事项&不推荐做法

JWT-TOKEN应该由服务端生成，永远不要让客户端生成TOKEN

有些项目中会把 签名密钥和算法名 交给调用方，由调用方来生成 JWT-TOKEN，这是一种 非常愚蠢的设计！！

这种做法有2个危害：

• 不安全。因为 签名密钥 泄露了，调用方可以生成各种权限的TOKEN，服务端根本没法限制或者校验。
• 数据结构没法升级。TOKEN中通常会包含一个数据结构（可参考前面示例），假设一开始 Token数据结构包含3个成员，和调用方约定后，人家按你的定义去实现了，你再想加其它成员就非常难了。
  （还有更离谱的真实案例）一开始的3个成员的赋值格式也可能不统一，典型的是 “有效期”这种时间格式，PHPer可能会给赋值一个数字~~ ，而你期望的是一个 DateTime

异步等待

虽然 .NET BCL 中提供了大量的异步API，但是有些时候，我们会遇到没有异步API的组件。

例如，某个第三方组件，它的内部执行过程比较耗时，它采用完成事件通知的方式来通知调用方。

示例代码如下：

public class Xjob {

	// 启动任务
	public void Start(string state){
		// 启动后台操作，有可能是交给硬件处理
		// 这个方法会立即返回
	}

    // 任务执行完成的事件通知
	public event EventHandler<OnCompletedArgs> OnCompleted;
}

对于这样一个组件，如果我们希望在 Web服务或者后台作业中调用，就非常麻烦，你可以自己思考下该如何实现~~~

为了解决这类问题，可以使用 ClownFish 提供的 ResultWaiter 来调用，示例代码如下：

public async Task<ActionResult> HttpAction1() {
	using ResultWaiter waiter = new ResultWaiter();

	Xjob xjob = new Xjob();
	xjob.OnCompleted += XjobOnCompleted;

	xjob.Start(waiter.ResultId);

	var result = await waiter.WaitAsync(TimeSpan.FromSeconds(30));
	// 处理 Xjob 的执行结果 ………………
}

private static void XjobOnCompleted(object sender, OnCompletedArgs e) {
     string resultId = e.State;

	 ResultWaiter waiter = ResultWaiter.GetById(resultId);
	 if( waiter != null ){
		if( e.Exception != null)
			waiter.SetException(e.Result);
		else
			waiter.SetResult(e.Result);
	 }
}

文本模板

文本模板可以理解为高级的string.Format(...)，它的用途是将 数据绑定到模板的占位标记 并生成一段新文本。

示例代码

private static readonly string s_httpTemplate = @"
POST http://www.abc.com/test/callback.aspx?data={enc(data.cn)}&v={rand} HTTP/1.1
x-header1: qqqqqqqqqqqqq
x-header2: wwwwwwwwwwwww
x-xx1: {data.xx1}
x-xx2: {enc(data.xx2)}
x-xx3: {data.xx333}
x-client-app: TestApp2
x-client-reqid: {guid}
x-client-reqid32: {guid32}
x-client-url: http://www.fish-test.com/aaa/bb.aspx
Content-Type: application/json; charset=utf-8
x-null: #{xx}#
xx-guid: {rand.guid}
xx-guid32: {rand.guid32}
xx-guid36: {rand.guid36}
xx-char: {rand.char}
xx-char0: {rand.char0}
xx-char1: {rand.char1}
xx-char2: {rand.char2}
xx-char3: {rand.char3}
xx-char4: {rand.char4}
xx-char5: {rand.char5}
xx-int: {rand.int}
xx-int0: {rand.int0}
xx-int1: {rand.int1}
xx-int2: {rand.int2}
xx-int5: {rand.int5}
xx-int9: {rand.int9}
xx-int10: {rand.int10}
xx-time1: {rand.now}
xx-time2: {5秒前}
xx-time3: {5分钟前}
xx-time4: {5小时前}
xx-time5: {5天前}
xx-time6: {2月前}
xx-value01: {rand}
xx-value02: {now}
xx-value03: {昨天}
xx-value04: {今天}
xx-value05: {明天}
xx-value06: {月初}
xx-value07: {下月初}
xx-value08: {季度初}
xx-value09: {下季度初}
xx-value10: {年初}
xx-value11: {明年初}
xx-value12: {周一}
xx-value13: {下周一}
xx-local1: {LocalSetting_key1}
xx-local2: {LocalSetting_key2}
xx-local3: {LocalSetting_keyxx}

{data}
";


public void Test1()
{
    object data = new { xx1 = 223, xx2 = "abcd", cn="中文汉字~!@#$#%" };

    TextTemplate template = new TextTemplate();
    string text = template.Populate(s_httpTemplate, data.ToDictionary());

    Console2.WriteLine(text);
}

输出结果

POST http://www.abc.com/test/callback.aspx?data=%e4%b8%ad%e6%96%87%e6%b1%89%e5%ad%97%7e!%40%23%24%23%25&v=637685405048053925 HTTP/1.1
x-header1: qqqqqqqqqqqqq
x-header2: wwwwwwwwwwwww
x-xx1: 223
x-xx2: abcd
x-xx3: {data.xx333}
x-client-app: TestApp2
x-client-reqid: 87d746f2-ec14-4fff-940a-477f61b1ca67
x-client-reqid32: 33de20e3ed014faeb976b3d16c00c83b
x-client-url: http://www.fish-test.com/aaa/bb.aspx
Content-Type: application/json; charset=utf-8
x-null: #{xx}#
xx-guid: 9a6e0f34-59e3-41ee-b21b-57033b25552a
xx-guid32: 7f57af257b2a4ef7ba3fe94d61a290c8
xx-guid36: 0c49a8e4-60ca-4038-bec9-48c99fdabd59
xx-char: b
xx-char0: {rand.char0}
xx-char1: 2
xx-char2: f1
xx-char3: 7da
xx-char4: ed84
xx-char5: 67466
xx-int: 63400309
xx-int0: {rand.int0}
xx-int1: 3
xx-int2: 39
xx-int5: 88929
xx-int9: 490415950
xx-int10: 1174402031
xx-time1: 2024-01-30 16:21:30
xx-time2: 2024-01-30 16:21:25
xx-time3: 2024-01-30 16:16:30
xx-time4: 2024-01-30 11:21:30
xx-time5: 2024-01-25 16:21:30
xx-time6: 2023-11-30 16:21:30
xx-value01: 638422284904214607
xx-value02: 2024-01-30 16:21:30
xx-value03: 2024-01-29
xx-value04: 2024-01-30
xx-value05: 2024-01-31
xx-value06: 2024-01-01
xx-value07: 2024-02-01
xx-value08: 2024-01-01
xx-value09: 2024-04-01
xx-value10: 2024-01-01
xx-value11: 2025-01-01
xx-value12: 2024-01-29
xx-value13: 2024-02-05
xx-local1: abcd
xx-local2: 1234
xx-local3: {LocalSetting_keyxx}

{"xx1":223,"xx2":"abcd","cn":"中文汉字~!@#$#%"}

模板语法

{xxx} 表示一个占位替换符，在调用Populate时【可能】会被替换。

{xxx}可以支持的范围

• {data} 表示整个数据对象的JSON序列化文本
• {data.name} 读取数据对象的 name 属性
• {rand} 获取一个随机数字
• {rand.name} 获取 XRandom 支持的一个随机值，具体可参考上面的示例
• {name} 同上，只是名称上简化了
• {enc(...)} 先计算括号内的数据（参考上面规则），再做 UrlEncode

如果匹配失败，占位符不做替换。

数据绑定

文本模板由TextTemplate工具类来实现，可以调用 Populate 来生成结果

public sealed class TextTemplate
{
	/// <summary>
	/// 填充模板
	/// </summary>
	/// <param name="template"></param>
	/// <param name="data"></param>
	/// <returns></returns>
	public string Populate(string template, IDictionary<string, object> data)


	/// <summary>
	/// 填充模板
	/// </summary>
	/// <param name="template"></param>
	/// <param name="json"></param>
	/// <returns></returns>
	public string Populate(string template, string json)

}

关键字替换

有些场景下，仅仅需要【关键字】的替换功能，例如： 在一个SQL语句中，写上固定的 占位符

SQL示例：

select * from table1 where create_time >= '{今天}' and create_time < '{明天}' and userid = '{USERID}'

上面这个SQL语句中就定义了3个关键词：

• 今天，例如：2023-01-01
• 明天，例如：2023-01-02
• USERID，例如：xxxxxxx

由于 XRandom 并不支持 {USERID} 这个【关键字】，所以需要自行实现一个，例如：

public static string GetCurrentUserId()
{
	// 例如（不能运行）：return HttpPipelineContext.Get().HttpContext.User.Identity.Name;
	return "u" + DateTime.Now.Ticks.ToString();
}

在【程序启动】时注册这个【关键字】，

XRandom.RegisterValueGetter("USERID", GetCurrentUserId);

然后，在业务代码中可调用：

string sql = XRandom.FillTemplate(sqltemplate);

即可得到下面类似的SQL语句：

select * from table1 where create_time >= '2024-01-30' and create_time < '2024-01-31' and userid = 'u638422290772964811'

请点击目录中的节点查看具体文档

HttpClient

本文将演示HttpClient的使用方法。

说明：

• 为了减少重复代码，本文只演示【同步】调用，例如：GetResult()
• 如果需要【异步】，请调用 Async 结尾的方法，例如：GetResultAsync()
  
  
  

GET请求

HttpOption httpOption = new HttpOption {
    Url = "http://www.fish-test.com/show-request2.aspx",
    Data = new { id = 2, name = "abc" }  // 这里的数据将会合并到URL中
};

string responseText = httpOption.GetResult();
Console.WriteLine(responseText);

或者

HttpOption httpOption = new HttpOption {
    Url = "http://www.fish-test.com/show-request2.aspx?id=2&name=abc",
};

string responseText = httpOption.GetResult();
Console.WriteLine(responseText);

POST 表单

string responseText = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-request2.aspx",
    Format = SerializeFormat.Form,
    Data = new { id = 2, name = "abc" }
}.GetResult();
Console.WriteLine(responseText);

POST 表单-含文件上传

string filename1 = "small_file.txt";    // 这里为了简单，使用了相对路径，实际使用时请使用绝对路径
string filename2 = "small_file.bin";
string filename3 = "ClownFish.Log.config";

string responseText = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-request2.aspx",
    Format = SerializeFormat.Multipart,  // 文件上传模式
    Data = new {
        id = 22222222222,
        name = "Fish Li",

        // 如果包含 FileInfo 或者 HttpFile 类型的属性值，就认为是上传文件
        file1 = new FileInfo(filename1),
        file2 = new FileInfo(filename2),
        file3 = new FileInfo(filename3),
    }
}.GetResult();
Console.WriteLine(responseText);

POST JSON

string responseText = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-body.aspx",
    Data = new { id = 2, name = "abc" },
    Format = SerializeFormat.Json     // 注意这里
}.GetResult();
Console.WriteLine(responseText);

POST Ndjson

List<Product2> list = CreateTestDataList(20);

string responseText = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-body.aspx",
    Data = list,
    Format = SerializeFormat.Ndjson     // 注意这里
}.GetResult();
Console.WriteLine(responseText);

Assert.IsTrue(responseText.Contains("Content-Type: application/x-ndjson"));

POST XML

string responseText = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-body.aspx",
    Data = new NameValue { Name = "abc", Value = "123" },
    Format = SerializeFormat.Xml     // 注意这里
}.GetResult();
Console.WriteLine(responseText);

POST 二进制数据

var data = new NameValue { Name = "abc", Value = "123" };
byte[] bytes = data.ToJson().ToUtf8Bytes();

string responseText = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-body.aspx",
    Data = bytes,
    Format = SerializeFormat.Binary     // 注意这里
}.GetResult();
Console.WriteLine(responseText);

POST 数据流

var data = new NameValue { Name = "abc", Value = "123" };
byte[] bytes = data.ToJson().ToUtf8Bytes();

MemoryStream ms = new MemoryStream(bytes);

string responseText = new HttpOption {
        Method = "POST",
        Url = "http://www.fish-test.com/show-body.aspx",
        Data = ms,
        Format = SerializeFormat.Binary     // 注意这里
    }.GetResult();
    Console.WriteLine(responseText);

提交压缩数据

var data = new NameValue { Name = "abc", Value = new string('x', 2048) };

HttpOption httpOption = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-request2.aspx",
    Data = data,
    Format = SerializeFormat.Json,
    AutoGzipUpload = true     // 注意这里
};

string text = httpOption.GetResult();
Console.WriteLine(text);

直接根据请求文本发送请求

// 下面这段文本，可以从 Fiddler 或者一些浏览器的开发工具中获取
// 拿到这段文本，不需要做任何处理，直接调用 HttpOption.FromRawText 就可以了，就是这样简单！


string request = @"
POST http://www.fish-test.com/show-request2.aspx HTTP/1.1
Host: www.fish-test.com
User-Agent: Mozilla/5.0 (Windows NT 6.3; WOW64; rv:36.0) Gecko/20100101 Firefox/36.0
Accept: */*
Accept-Language: zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3
Accept-Encoding: gzip, deflate
Content-Type: application/x-www-form-urlencoded; charset=UTF-8
X-Requested-With: XMLHttpRequest
Referer: http://www.fish-test.com/Pages/Demo/TestAutoFindAction.htm
Cookie: hasplmlang=_int_; LoginBy=productKey; PageStyle=Style2;
Connection: keep-alive
Pragma: no-cache
Cache-Control: no-cache

input=Fish+Li&Base64=%E8%BD%AC%E6%8D%A2%E6%88%90Base64%E7%BC%96%E7%A0%81
";

string responseText = 
    HttpOption.FromRawText(request)     // 构建 HttpOption 实例
    .GetResult();                       // 发送请求

Console.WriteLine(responseText);

结合RawText和Data发送请求

// 可以从 Fiddler 抓到所需要请求头，去掉：数据部分
string request = @"
POST http://www.fish-test.com/show-request2.aspx HTTP/1.1
Host: www.fish-test.com
User-Agent: Mozilla/5.0 (Windows NT 6.3; WOW64; rv:36.0) Gecko/20100101 Firefox/36.0
Accept: */*
Accept-Language: zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3
Accept-Encoding: gzip, deflate
X-Requested-With: XMLHttpRequest
Referer: http://www.fish-test.com/Pages/Demo/TestAutoFindAction.htm
Cookie: hasplmlang=_int_; LoginBy=productKey; PageStyle=Style2;
Connection: keep-alive
Pragma: no-cache
Cache-Control: no-cache
";

// 1，根据一段长文本 快速设置 URL, method, headers
HttpOption httpOption = HttpOption.FromRawText(request);

// 2，设置提交数据与格式
httpOption.Format = SerializeFormat.Form;
httpOption.Data = new { id = 2, name = "aaaa", time = DateTime.Now };

// 3，发送请求，获取结果
string result = httpOption.GetResult();
Console.WriteLine(result);

设置请求超时时间

string text = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-request.aspx",
    Data = new { id = 2, name = "abc" },
    Timeout = 3_000     // 3 秒超时
}.GetResult();
Console.WriteLine(text);



string text2 = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-request.aspx",
    Data = new { id = 2, name = "abc" },
    Timeout = 0     // 无限等待
}.GetResult();

发送时指定请求头

string text = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-request2.aspx",
    Data = new { id = 2, name = "abc" },
    Headers = new Dictionary<string, string>() {
                { "X-Requested-With", "XMLHttpRequest" },
                { "User-Agent", "Mozilla/5.0 (Windows NT 6.3; WOW64)"} }
}.GetResult();
Console.WriteLine(text);

处理请求头与响应头

HttpResult<string> result = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/test-header.aspx",
    Data = new { id = 2, name = "abc" },

    Header = new {      // 指定二个请求头，服务端会将它们合并
        X_a = "a1",     // 对应请求头：X-a: a1
        X_b = "b2"      // 对应请求头：X-b: b2
    }
}.GetResult<HttpResult<string>>();

// 注意调用上面方法时指定的泛型参数
// 如果需要读取响应头，需要指定 HttpResult<T> 的类型参数

// 读取响应结果
string responseText = result.Result;    // this is html

// 读取响应头
string header = result.Headers["X-add-result"];
Assert.AreEqual("a1b2", header);

Console.WriteLine(responseText);

发送请求时指定身份信息

string text = new HttpOption {
    Url = "http://RabbitHost:15672/api/queues",
    Credentials = new NetworkCredential("fishli", "abcde"),
    Timeout = 6_000,
}.GetResult();
Console.WriteLine(text);

发送当前进程的身份到服务端_服务端采用Windows身份认证

string text = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/show-request.aspx",
    Data = new { id = 2, name = "abc" },
    Credentials = CredentialCache.DefaultNetworkCredentials     // 注意这里
}.GetResult();
Console.WriteLine(text);

发送HTTP请求_维护会话COOKIE

// 准备一个Cookie会话容器
CookieContainer cookieContainer = new CookieContainer();

HttpOption page1 = new HttpOption {
    Url = "http://www.fish-test.com/user/page1.aspx",
};

// 没有登录，返回 403
int status = GetStatusCode(page1);
Assert.AreEqual(403, status);


// 登录
HttpOption login = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/user/login.aspx",
    Data = new { name = "fish li", pwd = "abc" },
    Cookie = cookieContainer // 接收cookie
};
string text = login.GetResult();
Console.WriteLine(text);
Assert.AreEqual("Login OK.", text);


// 再次访问，带上Cookie，就正常了
page1 = new HttpOption {
    Url = "http://www.fish-test.com/user/page1.aspx",
    Cookie = cookieContainer
};

int status2 = GetStatusCode(page1);
Assert.AreEqual(200, status2);

private static int GetStatusCode(HttpOption option)
{
    try {
        var xx = option.GetResult();
        return 200;
    }
    catch( RemoteWebException remoteWebException ) {
        return remoteWebException.StatusCode;
    }
    catch( WebException ex ) {
        return (int)(ex.Response as HttpWebResponse).StatusCode;
    }
    catch( Exception ) {
        return 500;
    }
}

获取各种类型的结果

// 先构造一个HttpOption对象
HttpOption httpOption = new HttpOption {
    Method = "POST",
    Url = "http://www.fish-test.com/test1.aspx",
    Header = new { X_a = "a1", X_b = "b2" },
    Format = SerializeFormat.Form,
    Data = new { id = 2, name = "abc" }
};


// 下面演示一些常见的获取结果的方式

// 1，以文本形式（含HTML）获取服务端的返回结果
string text = httpOption.GetResult();

// 2，以二进制形式获取服务端的返回结果
byte[] bin = httpOption.GetResult<byte[]>();

// 3，如果服务端返回 json / xml，可以直接通过反序列化得到强类型结果
Product product = httpOption.GetResult<Product>();

// 4，以文本形式获取服务端的返回结果，并需要访问响应头
HttpResult<string> httpResult1 = httpOption.GetResult<HttpResult<string>>();
string text2 = httpResult1.Result;
string header1 = httpResult1.Headers.Get("Content-Type");  // 读取响应头

// 5，以二进制形式获取服务端的返回结果，并需要访问响应头
HttpResult<byte[]> httpResult2 = httpOption.GetResult<HttpResult<byte[]>>();
byte[] bin2 = httpResult2.Result;

// 6，服务端返回 json / xml，结果反序列化，并需要访问响应头
HttpResult<Product> httpResult3 = httpOption.GetResult<HttpResult<Product>>();
Product product2 = httpResult3.Result;

// 7, 以Stream形式获取服务端的返回结果
// 注意：拿到结果后，请使用 using 包起来使用
Stream steram = httpOption.GetResult<Stream>();

// 8, 以HttpWebResponse形式获取服务端的返回结果
// 注意：拿到结果后，请使用 using 包起来使用
HttpWebResponse response = httpOption.GetResult<HttpWebResponse>();

仅仅发送请求，不需要读取结果

HttpOption httpOption = new HttpOption {
    Url = "http://www.fish-test.com/show-request2.aspx",
};

httpOption.Send();

发送请求_启用重试功能

HttpOption httpOption = new HttpOption {
    Url = "http://www.fish-test.com/show-request2.aspx",
};

// 创建默认的重试策略
// 当发生 “网络不通” 或者 “HTTP 502,503” 时，最大重试 7 次，每次间隔 1000 毫秒
Retry retry = HttpRetry.Create();
string html = httpOption.GetResult<string>(retry);

支持 unix-socket

例如调用 Docker API，curl 的命令是

curl --unix-socket /var/run/docker.sock http://localhost/v1.43/images/json

对应使用 HttpOption 的方式是：

HttpOption httpOption = new HttpOption {
    Url = "http://localhost/v1.43/images/json",
    UnixSocketEndPoint = "/var/run/docker.sock"
};
string json = httpOption.GetResult();

或者

string script = @"
GET http://localhost/v1.43/images/json HTTP/1.1
--unix-socket: /var/run/docker.sock
".Trim();

HttpOption httpOption = HttpOption.FromRawText(script);
string json = httpOption.GetResult();

Mail客户端

示例代码

public void SentTextMail()
{
    MailClient client = new MailClient("mail-config");
    client.SetReceivers("liqf01@mingyuanyun.com")
        .SetCC("liqf01@mingyuanyun.com")
        .SetSubject("MailClient_DEMO--TEXT---" + DateTime.Now.ToTimeString())
        .AddAttachment(@"files/asp.net.jpg", "image/jpeg")
        .SetBody("aaaaaaaaaaaaaaaaaa")
        .Send();
}


public void SendHtmlMail()
{
    string html = @"
<html>
<head><title>MailClient_DEMO</title></head>
<body><h1>MailClient_DEMO</h1><body>
</html>";

    MailClient client = new MailClient("mail-config");
    client.SetReceivers("liqf01@mingyuanyun.com")
        .SetCC("liqf01@mingyuanyun.com")
        .SetSubject("MailClient_DEMO--HTML---" + DateTime.Now.ToTimeString())
        .AddAttachment(@"files/asp.net.jpg", "image/jpeg")
        .SetHtmlBody(html)
        .Send();
}

指定收件人

指定【收件人】有3种方式，可参考下面示例代码：
【抄送人】的使用方式相同，所以没有特别指定。

client
    .SetReceivers("liqf01@mingyuanyun.com", "fangw@mingyuanyun.com", "yangmc@mingyuanyun.com", "412537239@qq.com")
    .SetSubject("Test MimeKit" + DateTime.Now.ToTimeString())
    .SetBody("aaaaaaaaaaaaaaaaaa")
    .Send();

client
    .SetReceivers("李奇峰1 <liqf01@mingyuanyun.com>", "方武2 <fangw@mingyuanyun.com>",
                  "杨敏超3 <yangmc@mingyuanyun.com>", "杨敏超4 <412537239@qq.com>")
    .SetSubject("Test MimeKit" + DateTime.Now.ToTimeString())
    .SetBody("aaaaaaaaaaaaaaaaaa")
    .Send();

client
    .SetReceivers( new NameValue("李奇峰1", "liqf01@mingyuanyun.com"), new NameValue("方武2", "fangw@mingyuanyun.com"), 
                   new NameValue("杨敏超3", "yangmc@mingyuanyun.com"), new NameValue("杨敏超4", "412537239@qq.com") )
    .SetSubject("Test MimeKit" + DateTime.Now.ToTimeString())
    .SetBody("aaaaaaaaaaaaaaaaaa")
    .Send();

SMTP服务器配置

在发送邮件前，请先在配置服务中注册SMTP服务器参数，可参考下图：

参数类型定义

public class SmtpConfig
{
    /// <summary>
    /// 服务器地址
    /// </summary>
    public string Host { get; set; }

    /// <summary>
    /// TCP端口
    /// </summary>
    public int Port { get; set; }

    /// <summary>
    /// 用户名
    /// </summary>
    public string UserName { get; set; }

    /// <summary>
    /// 密码
    /// </summary>
    public string Password { get; set; }

    /// <summary>
    /// 是否使用SSL连接
    /// </summary>
    public bool IsSSL { get; set; }
}

参数值配置示例

Host=smtp.exmail.qq.com;Port=587;UserName=xxx@mingyuanyun.com;Password=xxxxxx;IsSSL=1

IM客户端

ClownFish 对国内主流的IM提供了客户端支持可用于发送消息。
只需要一套代码即可支持3种IM

3种IM程序

• 企业微信
• 钉钉
• 飞书

针对3种IM，又支持3大类的使用场景：

• IM应用 给 单个用户 推送消息
• IM应用 给 聊天群 推送消息
• 以IM-Webhook 方式给 聊天群 推送消息

IM应用的连接认证配置

在发送消息前，请先在配置服务中注册IM所需的连接认证参数，可参考下图：

参数类型定义

/// <summary>
/// 企业微信/钉钉/飞书 应用 登录参数
/// </summary>
public sealed class ImAppAuthConfig
{
	/// <summary>
	/// IM类别
	/// </summary>
	public ImType ImType { get; set; }

	/// <summary>
	/// 企业微信的 CorpId，飞书的 AppID，钉钉的 AppKey
	/// </summary>
	public string AppId { get; set; }

	/// <summary>
	/// 企业微信的 Secret，飞书的 AppSecret，钉钉的 AppSecret
	/// </summary>
	public string AppSecret { get; set; }

	/// <summary>
	/// 企业微信/钉钉的 AgentId，飞书不使用此参数
	/// </summary>
	public long AgentId { get; set; }
}

public enum ImType
{
	WxWork,
	DingDing,
	FeiShu
}

参数值配置示例

ImType=WxWork;AgentId=111111;AppId=xxxxxxxxxxxx;AppSecret=xxxxxxxxxxxxxxxx

IM应用给单个用户推送消息

示例代码

public static async Task Test1()
{
	string userId = "liqf01";
	ImAppMsgClient client = new ImAppMsgClient("WxWork.AppAuth.Config");
	
	await client.SendTextAsync(userId, TextMsg);
	await client.SendMarkdownAsync(userId, MarkdownMsg);
	await client.SendImageAsync(userId, @"e:\aaaaa.jpg");

	await client.SendFileAsync(userId, @"E:\bbbbb.pdf");
	await client.SendFileAsync(userId, @"E:\ccccc.txt");
	await client.SendFileAsync(userId, @"E:\ddddd.png");
	
	await client.SendCardAsync(userId, CardTitle, CardMsg, CardHref);
}

IM应用给聊天群推送消息

示例代码

public static async Task Test2()
{
    string chatId = "xxxxxxxxxxxxxxxxxxxxxxxxxxx";	
	ImGroupChatClient client = new ImGroupChatClient("WxWork.AppAuth.Config");	

	await client.SendTextAsync(chatId, TextMsg);
	await client.SendMarkdownAsync(chatId, MarkdownMsg);
	await client.SendImageAsync(chatId, @"e:\aaaaa.jpg");

	await client.SendFileAsync(chatId, @"E:\bbbbb.pdf");
	await client.SendFileAsync(chatId, @"E:\ccccc.txt");
	await client.SendFileAsync(chatId, @"E:\ddddd.png");
	
	await client.SendCardAsync(chatId, CardTitle, CardMsg, CardHref);
}

注意事项

• A应用创建的“XX群”，B应用不能给“XX群”发消息
• A应用要创建群，需要在企业微信后台的“可见范围”中设置必需的可见范围

以IM-Webhook方式给聊天群推送消息

示例代码

public static async Task Test1()
{
	ImWebhookClient client = new ImWebhookClient("WxWork.WebHookAuth.Config");
	
	await client.SendTextAsync(TextMsg);
	await client.SendMarkdownAsync(MarkdownMsg);
}

WebHook配置参数类型定义

/// <summary>
/// 企业微信/钉钉/飞书 WebHook 登录参数
/// </summary>
public sealed class ImWebHookConfig
{
	/// <summary>
	/// IM类别
	/// </summary>
	public ImType ImType { get; set; }

	/// <summary>
	/// WebHook URL
	/// </summary>
	public string WebHookUrl { get; set; }

	/// <summary>
	/// 签名密钥，可为空。
	/// </summary>
	public string SignKey { get; set; }
}

企业微信专用客户端

如果需要调用企业微信的服务端API，可以使用 WxworkClient 这个类型。

例如，获取用户信息：

WxworkClient client = new WxworkClient("WxWork.AppAuth.Config");

WxworkUserInfo userinfo = client.GetUserInfo("liqf01");

Console2.WriteLine(userinfo.ToJson(JsonStyle.Indented));

其它有用的方法

• public T SendRequest(HttpOption httpOption) where T : IShitResult
• public async Task SendRequestAsync(HttpOption httpOption) where T : IShitResult
• public T SendData(object data) where T : IShitResult
• public async Task SendDataAsync(object data) where T : IShitResult
• public string UploadMedia(byte[] fileBody, string fileName, string mediaType)
• public async Task UploadMediaAsync(byte[] fileBody, string fileName, string mediaType)

使用WxworkClient的好处：

• 它会在内部自动处理企业微信所需的 access_token
• 支持开发阶段查看完整的 请求/响应 过程，环境变量：Nebula_ImHttpClient_Debug_Enabled=1

例如：

使用 InfluxDB

在配置服务中注册连接

在使用 InfluxDB 前，请先在配置服务中注册连接参数，可参考下图：

上图注册了一个连接，名称是 InfluxDBTest

创建客户端实例

根据“连接名称”创建客户端实例 （示例代码使用了上图中的连接名称）

InfluxClient client = InfluxClient.Create("InfluxDBTest");

对于SaaS场景，可以采用下面的方式创建InfluxClient实例

InfluxClient client = InfluxClient.CreateTenant("tenantId");

写入数据到 InfluxDB

public static async Task Test_Write()
{
	// 生成要写入的测试数据
	List<RequestCount> list = CreateTestData();

	// 创建InfluxDB的客户端实例，Create方法的参数是一个“连接名称”，它已在配置服务中注册为数据库连接
	InfluxClient client = InfluxClient.Create("InfluxDBTest");

	// 执行插入操作
	await client.InsertAsync(list);
}

示例代码中数据结构的定义

public sealed class RequestCount
{
	[TagField]
	public string AppName { get; set; }
	[TagField]
	public string HostName { get; set; }

	[ValueField]
	public int Count { get; set; }

	[ValueField]
	public int Error { get; set; }

	[ValueField]
	public int Slow { get; set; }

	[ValueField]
	public int AvgTime { get; set; }

	[TimeField]
	public DateTime CreateTime { get; set; }
}

从 InfluxDB 中查询数据

示例-1--无参数

public static async Task Test_Query()
{
	// 查询最近 6 小时数据，30分钟为一个采样频率
	string query = @$"select sum(Count) as Count, sum(Error) as Error from RequestCount 
				WHERE AppName='TestApp0' and time > now() - 6h group by time( 30m )";

	// 执行查询
	InfluxClient client = InfluxClient.Create("InfluxDBTest");
	DataTable table = await client.ToTableAsync(query);

	Console.WriteLine(table.ToJson(JsonStyle.Indented));
}

示例-2--参数化查询

public static async Task Test_Query2()
{
	DateTime today = DateTime.Today;
	long start = today.GetNanoTime();
	long end = today.AddHours(6).GetNanoTime();
	
	// 带有占位参数的查询， $start, $end， 1个小时为一个采样频率
	string query = @$"select sum(Count) as Count, sum(Error) as Error from RequestCount 
				WHERE time >= $start and time < $end group by time( 1h )";

	var args = new { start, end };  // 查询参数

	// 执行查询
	InfluxClient client = InfluxClient.Create("InfluxDBTest");
	DataTable table = await client.ToTableAsync(query, args);

	Console.WriteLine(table.ToJson(JsonStyle.Indented));
}

下面这些InfluxDB查询都是有效的

select * from RequestCount 
WHERE AppName='TestApp0'  ORDER BY time DESC LIMIT 300

select * from RequestCount 
WHERE time >= '2020-10-15 00:00:00' and time < '2020-10-16 00:00:00' ORDER BY time DESC LIMIT 300

select sum(Count) as Count, sum(Error) as Error from RequestCount 
WHERE time >= '2020-10-15 00:00:00' and time < '2020-10-16 00:00:00' and AppName='TestApp0' group by time(1h)

select sum(Count) as Count, sum(Error) as Error from RequestCount 
WHERE AppName='TestApp0' and time > now() - 24h group by time(1h)

InfluxDB查询语法参考链接： https://docs.influxdata.com/influxdb/v1.8/query_language/

在查询服务中使用 InfluxDB

在查询服务中创建服务接口可参考下图：

说明：

• SQL中的参数名以 $ 开头
• 如果需要查看 InfluxDB 的原始响应，调用时请指定：x-result-format=raw

数据查询服务支持以下返回格式：

• json (默认值)
• xml
• xml2
• csv
• excel
• raw

使用 VictoriaMetrics

在配置服务中注册连接

在使用 VictoriaMetrics 前，请先在配置服务中注册连接参数，可参考下图：

上图注册了一个连接，名称是 vm-conn

创建客户端实例

根据“连接名称”创建客户端实例 （示例代码使用了上图中的连接名称）

VictoriaMetricsClient client = VictoriaMetricsClient.Create("vm-conn");

对于SaaS场景，可以采用下面的方式创建VictoriaMetricsClient实例

VictoriaMetricsClient client = VictoriaMetricsClient.CreateTenant("tenantId");

写入数据到 VictoriaMetrics

public static async Task Test_Write()
{
	// 生成要写入的测试数据
	List<RequestCount> list = CreateTestData();

	// 创建VictoriaMetrics的客户端实例，Create方法的参数是一个“连接名称”，它已在配置服务中注册为数据库连接
	VictoriaMetricsClient client = VictoriaMetricsClient.Create("vm-conn");

	// 将测试数据写入VictoriaMetrics
	await client.GetInsertRequest(list).SendAsync();
}

示例代码中数据结构的定义

public sealed class RequestCount
{
	[TagField]
	public string AppName { get; set; }
	[TagField]
	public string HostName { get; set; }

	[ValueField]
	public int Count { get; set; }

	[ValueField]
	public int Error { get; set; }

	[ValueField]
	public int Slow { get; set; }

	[ValueField]
	public int AvgTime { get; set; }

	[TimeField]
	public DateTime CreateTime { get; set; }
}

从 VictoriaMetrics 中查询数据

public static async Task Test_Query()
{
	// VictoriaMetrics 使用的 PromQL 查询
	string query = "RequestCount:Count";
	// 3个查询过滤参数
	DateTime start = DateTime.Now.AddHours(-3);
	DateTime end = DateTime.Now;
    int step = 0;

	// 创建VictoriaMetrics的客户端实例，Create方法的参数是一个“连接名称”，它已在配置服务中注册为数据库连接
	VictoriaMetricsClient client = VictoriaMetricsClient.Create("vm-conn");
	
	// 执行查询，此时返回的结果是JSON格式
	string json = await client.GetQueryRequest(query, start, end, step).GetResultAsync();

	// 将JSON数据转成 DataTable
	DataTable table = client.JsonToDataTable(json);

	Console.WriteLine(table.ToJson(JsonStyle.Indented));
}

说明：

• VictoriaMetrics的返回结果只有一个数值列
• JSON转DataTable时，会固定生成 XTime, XValue 这二个列，表示时间和数值

在查询服务中使用 VictoriaMetrics

在查询服务中创建服务接口可参考下图：

说明：

• 数据源：一定要选择 VictoriaMetrics，连接也要匹配
• 查询脚本：就是 PromQL 查询
• StartTime,EndTime,Step 这3个参数是必须的，用于过滤数据
• 由于PromQL并不支持参数化查询，所以 @db 只是一个占位标记，在执行时会被替换

数据查询服务支持以下返回格式：

• json (默认值)
• xml
• xml2
• csv
• excel
• raw

查询演示

使用 ElasticSearch

在配置服务中注册连接

在使用 ElasticSearch 前，请先在配置服务中注册连接参数，可参考下图：

上图注册了一个连接，名称是 ClownFish_Log_Elasticsearch

创建客户端实例

根据“连接名称”创建客户端实例 （示例代码使用了上图中的连接名称）

EsConnOption option = EsConnOption.Create("ClownFish_Log_Elasticsearch");
SimpleEsClient client = new SimpleEsClient(option);

写入数据到 ElasticSearch

// 写入一个对象
NotifySendLog log = new NotifySendLog {.....};
await client.WriteOneAsync(log);

// 批量写入
List<NotifySendLog> list = .....;
await client.WriteListAsync(list);

数据在写入ES时，创建的索引名为：classname-yyyyMMdd，例如：notifysendlog-20220408

请点击目录中的节点查看具体文档

HTTP请求转队列

为了能应对客户端的大量数据提交请求，可以采用队列来缓冲请求压力，具体有2个步骤：

• 将整个HttpRequest存入队列
• 在另一个进程中订阅队列消息来处理HttpRequest

HttpRequest存入队列

[HttpPost]
public async Task<int> Databus()
{
    // 将请求序列化字节数组
    byte[] data = await this.RequestToBytesWithMonitorAsync();

    // 将当前请求发送到MQ
    this.SendRabbitMessage(ResNames.Rabbit, data, s_queueName);

    return data.Length;
}

如果需要将当前请求发到不同的MQ，请按下面方式处理：

[HttpPost]
public async Task<int> Databus()
{
    // 将请求序列化字节数组
    byte[] data = await this.RequestToBytesWithMonitorAsync();

    // 如果需要将当前请求发到不同的MQ，请按下面方式处理：
    if( /* 条件1 */ true )
        this.SendRabbitMessage(ResNames.Rabbit, data, "queue_1111");

    if( /* 条件2 */ true )
        this.SendRabbitMessage(ResNames.Rabbit, data, "queue_2222");

    if( /* 条件3 */ true )
        this.SendRabbitMessage(ResNames.Rabbit, data, "queue_3333");

    return data.Length;
}

处理HttpRequest

public class RequestMsgHandler : BaseMessageHandler<NHttpRequest>
{
    public override void ProcessMessage(PipelineContext<NHttpRequest> context)
    {
        NHttpRequest request = context.MessageData;

        // 在这里可以获取到从MQ中还原出来的 request 对象，你可以像 Controller中那样处理
        // 好处是，这里没有并发压力

        //// 例如：读取请求头
        //string contentType = request.ContentType;
        //string datatype = request.Header("x-datatype");

        //// 获取请求体内容
        //string body = request.BodyText;  // 注意：这里已经做过了 gzip 解压缩

        //// 获取客户端发送的 List 对象
        //List<OprLog> list = body.FromNdjson<OprLog>();

        //foreach(var x in list ) {
        //    // 处理逻辑…………
        //}

    }
}

客户端上传数据与服务端队列处理

客户端上传数据

要求

• 数据包不能无限大，服务端只允许1M大小
• 数据上传前做 gzip 压缩

参考示例

这里假设采集到的数据是一个 List 集合

代码解释：

• spliter.GetNextPart() 的返回结果是一个 多行JSON，不是JSON数组
• 返回结果是一个字符串，长度不超过 6M（参数）
• 在发送前调用 ToGzip 方法可以实现数据压缩，可以控制在 1M 内

发送数据到服务端：

代码解释：

• x-datatype ： 可用于服务端的兼容性处理（提示作用）
• Content-Encoding = gzip ：服务端就知道这是个 gzip 压缩数据
• ContentType.Binary：指示请求体是一个二进制数据，避免一些服务端模块错误处理

云端入口服务

这里采用队列来缓冲，防止大量请求突然出现而丢失数据包。

示例代码说明：

• Action方法中可以不做作任何处理，直接把 request 发到MQ就可以了
• request在发到MQ时，数据不做解压缩，不做反序列化，能保持数据包在一个较小的长度

订阅队列消息

处理消息

兼容性处理示例

分析总结

• 借助 DataSpliter 的分割功能，数据包大小可控
• 上传时使用 gzip 压缩，数据包减少至少8倍
• 整个过程只有一次JSON序列化和一次反序列化
• 序列化是一个对象一个对象的进行，反序列化时不会占用大量内存
• 整个过程只有一次 gzip 压缩 和 解压缩

HTTP转发代理

使用场景：

• 少部分InternalService的服务接口需要通过PublicService对外公开，此时就需要在PublicService收到请求时转发到InternalService
  
  

新建一个PublicService

核心步骤就是在启动时调用 AppStartup.RunAsPublicServices(...) 方法，例如：

public class Program
{
    public static void Main(string[] args)
    {
        AppStartup.RunAsPublicServices("Nebula.Demo", args);
    }
}

新建配置文件

在应用程序目录下创建一个名为 Nebula.Demo.TransferMapRule.xml 的配置文件。

请注意转发配置文件的命名规则 ：

• 前缀，当前程序的名称，在这个示例中：Nebula.Demo
• 后缀，固定为：.TransferMapRule.xml

配置文件内容示例如下：

<ProxyMapRule>
    <Rules>
        <rule src="/new/api/test/add"           dest="{service1_Url}/v20/api/WebSiteApp/test/Add.aspx" />
        <rule src="/new/api/test/ShowRequest"   dest="{service1_Url}/v20/api/WebSiteApp/test/ShowRequest.aspx" />
        <rule src="/new/api/test/login"         dest="{service2_Url}/v20/api/WebSiteApp/auth/login.aspx" />
        <rule src="/new/api/test/findByTel"     dest="{service2_Url}/v20/api/WebSiteApp/Customer/findByTel.aspx" />

        <rule src="/v20/api/moon/[any]"         dest="{configServiceUrl}/v20/api/moon/[any]" />
    </Rules>
</ProxyMapRule>

然后设置文件属性，参考下图：

规则说明

• 所有规则从上到下依次匹配，遇到匹配的规则后立即执行
  
  
• src 包含 [any] 时
  • src 表示要执行匹配的URL模式
    • [any] 表示一个占位符，可以匹配任何多个字符
  • dest 表示将要转发的目标地址，它必须是一个绝对地址
    • dest也必须包含 [any]，用于计算转发目标，它的值从 src的[any]中获得
  • 实际转发请求时，dest_url = dest + request.QueryString
    
    
• src 不 包含 [any] 时
  • src 表示要执行匹配的URL路径
  • dest 表示将要转发的目标地址，它必须是一个绝对地址
  • 实际转发请求时，url = dest + request.Query
    
    
• src = * 时
  • 表示匹配当前站点所有请求，所以如果需要这个规则，一定要放在最后。
  • dest 是一个URL路径前缀
  • 实际转发请求时，url = dest + request.PathAndQuery

举例说明

假设当前站点的根地址：http://www.abc.com/

src = "/new/api/test/add",
dest = "http://localhost2:8206/v20/api/WebSiteApp/test/Add.aspx"

示例请求：http://www.abc.com/new/api/test/add?id=2
转发目标：http://localhost2:8206/v20/api/WebSiteApp/test/Add.aspx?id=2

src = "/v20/api/moon/[any]",
dest = "http://LinuxTest:8503/v20/api/moon/[any]"

示例请求：http://www.abc.com/v20/api/moon/setting/get.svc?name=xxx
转发目标：http://LinuxTest:8503/v20/api/moon/setting/get.svc?name=xxx

WebHook事件

WebHook的用途有2点：

• 应用解耦：应用程序产生WebHook事件时，不用关心有多少个订阅者程序
• 应用集成：可通过订阅事件的方式实现数据同步

WebHook运行示意图

可以发现

• MyApp1 与 OtherApp1/2/3/4 不直接调用（没有耦合）
• Event1 有 3个订阅者（MyApp1不用关心）
• Event2 有 2个订阅者（MyApp1不用关心）

总结：

• WebHook事件的发起方与订阅方完全解耦，它们只和WebHookServer这个中间交互。
• 一个WebHook事件可以有多个订阅者，也可以没有任何订阅者，对于这些发起方不用关心
• 发起方只需要一套代码即可整合多个其它应用程序

发送WebHook事件

// 创建事件数据对象
var alert = new {
	AlertId = 123,
	Message = "xxxxxxxxxxx",
	AlertType = 11,
	// ...............
};

// 发布WebHook事件
WebHookClient.Instance.PublishEvent("alert.create", alert);

WebHookServer

Nebula自带一个WebHookServer的实现，
具体内容可参考： WebHook服务

全局锁（分布式锁）

使用场景：在多进程之间保证资源只能由一个进程访问。

示例代码

/// <summary>
/// 多队列-多进程 自动配对订阅实现方案，
/// 假如有 10 个队列，5个进程，此时每个进程将订阅2个队列
/// </summary>
private static void Demo()
{
    // 此方案的大致思路是：
    // 1，先获取一个分布式锁，保证多个进程互斥（只有一个进程能进入执行）
    // 2，获取所有队列信息，检查将要订阅的队列是否有订阅者，如果队列没有订阅者，就订阅它


    // 创建一个分布式锁，锁将保证多个进程时，只有一个进程能进入执行
    using( GlobalLock locker = GlobalLock.Create("一个特殊的锁标识") ) {

        RabbitMonitorClient client = new RabbitMonitorClient("rabbitmq-连接名称");

        // 获取所有队列信息，并过滤【特定】
        List<QueueInfo> list = client.GetQueues(10_000).Where(x => x.Name.StartsWith("队列名称前缀")).ToList();

        // 记录当前进程已订阅的队列数量
        int count = 0;

        foreach( var q in list ) {

            // 如果某个队列没有订阅者，就表示当前进程可以订阅它
            if( q.Consumers == 0 && count < 2 ) {

                // 订阅某个队列
                RabbitSubscriberArgs args = new RabbitSubscriberArgs {
                    SettingName = "Rabbit连接配置名称",
                    QueueName = q.Name,
                    SubscriberCount = 3
                };
                RabbitSubscriber.Start<XMessage, XMessageHanlder>(args);
                count++;
            }
        }
    }
}

在上面这个示例中，受保护的资源就是RabbitMQ，

为了实现多个进程节点 均匀且不重复 订阅，就需要将RabbitMQ保护起来，

一次只允许一个进程访问RabbitMQ的队列信息，并从中选择2个没有被订阅的队列。

跨进程事件通知

跨进程事件通知 不同于简单的HTTP调用通知：它是一种一对多的通知方式。

典型场景:

• 数据修改后通知【多个进程】及时刷新内存中缓存

发布事件通知

AppEvent.PublishGlobalEvent("insert.userservice.userrole", "usercode");

AppEvent.PublishGlobalEvent("update.userservice.userrole", "123", "usercode");

PublishGlobalEvent方法签名

/// <summary>
/// 发送【跨进程】事件，事件以异步方式交给EventHandler执行。
/// </summary>
/// <param name="name">事件名称</param>
/// <param name="param1">参数1</param>
/// <param name="param2">参数2</param>
/// <param name="param3">参数3</param>
public static void PublishGlobalEvent(string name, string param1 = null, string param2 = null, string param3 = null)

为什么是3个参数(param1/param2/param3)？

设计原因：

• 使用简单：大多数简单场景可以直接使用简单数据类型(int/string)
  • 例如：缓存更新时，可以只指定 param1=cacheKey 参数即可
• 方便过滤：减少反序列次数
  • 由于通知会发送给集群内所有进程，为了便于过滤，可以将过滤条件赋值给 param2/param3

说明：

• 以上只是使用建议，你可以给param1/param2/param3赋值任意复杂的JSON文本，甚至是null
• 这3个参数具体表达什么含意，由每个事件来决定！

订阅&处理事件

public class XxxEventHandler : BaseAppEventHandler
{
     [EventAction(EventName = "insert.userservice.userrole")]
     public void Method1(AppEventArgs e)
     {
          // 事件处理逻辑
     }


     [EventAction(EventName = "update.userservice.userrole")]
     public async Task Method2(AppEventArgs e)
     {
      // 事件处理逻辑
     }
}

注意事项：

• 事件订阅类型必须定义成 public ，且必须从 BaseAppEventHandler 继承
• 订阅事件的方法必须定义成 public
• 方法的返回值类型只能是 void 或者是 Task
• 方法只能包含一个参数，类型是 AppEventArgs
• 方法必须用 [EventAction] 来明确指定要订阅的事件名称

如何忽略某个事件？

方法：

• 不订阅那个事件

说明：

• 事件处理方法是异步调用的，不会阻塞发布方，也没有返回值
• 不论事件如何处理，对发布方是没有影响的

与消息处理相比

【跨进程事件通知】与【消息处理】其实是非常类似的设计。

跨进程事件通知 的主要设计目标是：

• 轻量化
• 易用使用

注意事项

基于上表描述，跨进程事件通知只能用于【比较简单】的通知场景，且在使用时应该注意：

• 事件会发送所集群内的所有进程，所以请优先考虑使用MQ的方式解决
• 事件处理方法必须要快速完成，否则当出现大量事件时，会导致事件丢失
• 默认的异常处理仅仅是 Console.Write，如果需要记录到异常日志，请手动处理

使用 ndjson

什么是 ndjson ？ (此段内容来自于网络)

ndjson 是一种 MIME 类型，用于表示 Newline Delimited JSON 格式的数据。

ndjson 是一种方便的格式，用于存储或流式传输结构化数据，每条记录可以逐行处理。

ndjson 的优势

• 流式处理：ndjson 使整个文件“流化”，将文件分割成许多份，避免了整体的束缚，支持局部处理，变得更灵活更快。
• 性能提升：在处理大数据集时，ndjson 格式可以显著提高性能，因为它允许逐行处理数据，而不需要一次性加载整个文件。
• 简化并行处理：由于每行都是一个独立的 JSON 对象，ndjson 格式的数据可以很容易地分割并行处理。

ndjson 和 json 的差异

传统的JSON数组

[
  {"Name":"x1","Value":"aaa"}, {"Name":"x2","Value":"bbb"},
  {"Name":"x3","Value":"ccc"}, {"Name":"x4","Value":"ddd"},
  {"Name":"x5","Value":"eee"}
]

等效的 ndjson 格式

{"Name":"x1","Value":"aaa"}
{"Name":"x2","Value":"bbb"}
{"Name":"x3","Value":"ccc"}
{"Name":"x4","Value":"ddd"}
{"Name":"x5","Value":"eee"}

最直观的差别有2点：

• 没有一对中括号 []
• 一行一个json对象，用换行符分隔，不需要逗号

ndjson 的用途

由于没有一对 [] 的限制，并且一行一个json，它就非常容易能形成【数据流】，

再借助 .NET Stream API 的强大功能，就可以实现：只需要很小的缓冲区就能处理无限大的数据量

反之传统的 json 数组，它的限制在于：所有数据必须全部放在内存中，内存消耗较大。

ndjson 的典型使用场景

1. 将数据库的查询结果输出成 ndjson，写入文件或者输出到网络流
2. 服务端返回列表数据，以ndjson形式输出，客户端用流的方式逐行读取并处理
3. 客户端发送列表数据，以ndjson形式上传，服务端用流的方式逐行读取并处理

下面通过几个示例来展示 ndjson 的使用方法，共分为2大类：

• 小数据量：可用于代替 json-array
• 无限数据量：ndjson 的优势体现

示例展示-1--SQL查询导出成ndjson并gzip压缩写入文件--无限数据量

using DbContext dbContext = DbContext.Create("connection_name");
await dbContext.OpenConnectionAsync();
dbContext.BeginTransaction(IsolationLevel.ReadUncommitted);  // 建议导出数据允许脏读

CPQuery query = dbContext.CPQuery.Create("select * from xxxxx", args);

using( FileStream fileStream = File.OpenWrite("d:/big-query-gzip-data.ndjson") ) {
	using( GZipStream gZipStream = new GZipStream(fileStream, CompressionMode.Compress, true) ) {
		using( StreamWriter writer = new StreamWriter(gZipStream, EncodingUtils.UTF8NoBOM, 4096, true) ) {

		int count = await query.ExportToNdJsonAsync(maxRows: -1, writer);   // 执行SQL查询并导出数据
			Console.WriteLine($"SQL查询执行完成，row-count: {count}");
		}
	}
}

示例展示-2--服务端以ndjson形式返回数据--小数据量

[HttpGet]
[Route("table/list")]
public NdjsonResult ClientList()
{
	List<ClientChannel> list = ClientListUtils.GetList();
	list = FilterList(list);

	return new NdjsonResult(list);    // 返回【小量级】数据
}

浏览器收到的响应头如下：

说明：NdjsonResult 在输出时默认会开启 gzip 压缩。

示例展示-3--服务端以ndjson形式返回数据--无限数据量

如果需要返回的数据量非常大，使用NdjsonResult就不合适了，

这里结合前面的示例来实现一个更完整的数据导出功能：

[HttpPost]
[Route("export/data/ndjson")]
public async Task DataExport()
{
	using DbContext dbContext = DbContext.Create("connection_name");
	await dbContext.OpenConnectionAsync();
	dbContext.BeginTransaction(IsolationLevel.ReadUncommitted);  // 建议导出数据允许脏读

	CPQuery query = dbContext.CPQuery.Create("select * from xxxxx", args);

	using( TempFileStream fileStream = new TempFileStream() ) {  // 这里用临时文件来缓冲数据

		using( GZipStream gZipStream = new GZipStream(fileStream, CompressionMode.Compress, true) ) {
			using( StreamWriter writer = new StreamWriter(gZipStream, EncodingUtils.UTF8NoBOM, 4096, true) ) {

				int count = await query.ExportToNdJsonAsync(maxRows: -1, writer);   // 执行SQL查询并导出数据

				dbContext.Dispose();   // ############## 提前关闭数据库连接
			}
		}

		this.NHttpContext.Response.SetHeader(HttpHeaders.Response.ContentEncoding, "gzip");

		// 直接将临时文件流复制到响应流
		await this.NHttpContext.HttpReplyAsync(200, fileStream, ResponseContentType.Ndjson);
	}
	// 临时文件会在此处自动删除
}

示例展示-4--客户端接收ndjson数据并转成强类型对象--小数据量

[TestMethod]
public async Task Test2()
{
	HttpOption httpOption = new HttpOption {
		Url = "http://linuxtest:8208/v20/api/fides/data/table/list"
	};

	List<EndClientUserInfo> list = await httpOption.GetResultAsync<List<EndClientUserInfo>>();

	Assert.IsTrue(list.Count > 0);
}

HTTP客户端自动识别了2个响应头

• Content-Type: application/x-ndjson
• Content-Encoding: gzip

示例展示-5--客户端接收ndjson数据并转成强类型对象--无限数据量

[TestMethod]
public async Task Test3()
{
	HttpOption httpOption = new HttpOption {
		Url = "http://linuxtest:8208/v20/api/fides/data/table/list"
	};

	HttpResult<Stream> httpResult = await httpOption.GetResultAsync<HttpResult<Stream>>();

	int count = 0;
	using NdJsonReader reader = NdJsonReader.Create(httpResult);

	foreach(var item in reader.ReadLines<EndClientUserInfo>() ) {
		count++;
	}

	Assert.IsTrue(count > 0);
}

示例展示-6-客户端以ndjson方式上传数据--小数据量

// 数据量不大，全部在内存中
List<Product3> list = Product3.CreateTestDataList(20);

HttpOption httpOption = new HttpOption {
	Method = "POST",
	Url = "http://www.fish-test.com/show-body.aspx",
	Data = list,
	Format = SerializeFormat.Ndjson     // 注意这里
};

await httpOption.SendAsync();

示例展示-7-客户端以ndjson方式上传数据--无限数据量

将一个 流 对象中的数据发送到服务端，这里使用前面示例产生的临时文件

// 数据量比较大，用文件来缓存，这里使用 示例1 产生的文件
using FileStream fileStream = File.OpenRead("d:/big-query-gzip-data.ndjson");

HttpOption httpOption = new HttpOption {
	Method = "POST",
	Url = "http://www.fish-test.com/show-body.aspx",
	Data = fileStream,      // 这里直接使用流对象
	Format = SerializeFormat.Ndjson,
	Header = new {
		Content_Encoding = "gzip"  // 写文件时使用了gzip压缩，所以这里需要指定
	},
	Timeout = 60_000
};
await httpOption.SendAsync();

示例展示-8--服务端从HTTP请求流中读取ndjson--小数据量

[HttpPost]
[Route("import/data/ndjson")]
public async Task ImportData()
{
	List<Product3> list = this.NHttpContext.Request.ReadBodyAsNdjonsToList<Product3>();

    using DbContext dbContext = DbContext.Create("connection_name");

	foreach( var item in list ) {
		await dbContext.Entity.InsertAsync(item);
	}
}

示例展示-9--服务端从HTTP请求流中读取ndjson--无限数据量

[HttpPost]
[Route("import/data/ndjson")]
public async Task ImportData()
{
	using NdJsonReader reader = NdJsonReader.Create(this.NHttpContext.Request);

	using DbContext dbContext = DbContext.Create("connection_name");

	foreach( var item in reader.ReadLines<Product3>() ) {
		await dbContext.Entity.InsertAsync(item);
	}
}

请点击目录中的节点查看具体文档

应用程序初始化

如果应用程序需要在业务处理前完成一些初始化动作，

例如：程序使用了RabbitMQ队列，我们希望在程序启动时就创建好队列，

诸如这类场景可以参考本文介绍的方法。

开发过程

• 在项目中新建一个名为 AppInitializer 静态类
• 增加一个名为 Init 的方法
• 在 Init 方法中编写初始化逻辑

示例代码

namespace YourProjectNameSpace;

public static class AppInitializer
{
    public static void Init()
    {
        // 在这里编写初始化逻辑
    }
}

补充说明：

• 示例代码中的类名和方法名是固定的，不能随意取名
• 示例代码中的类型和方法都要定义成静态的
• Init方法不需要在代码中调用，Nebula会在启动时调用

注意：

• 不要直接在Program.Main方法中执行初始化操作

类库项目初始化

在类库项目中，AppInitializer类不会纳入查找范围，所以即使有定义也不会被调用。

此时可以参照以下方法：

[Init]  // 指示这个类需要在启动时初始化
public static class ComponentInitializer
{
    public static void Init()
    {
        Console.WriteLine("Demo.Data.ComponentInitializer.Init() called!");
    }
}

说明：

• 类型的可见性必须是 public
• 方法的签名必须是：public static void Init()

Nebula行为定制

Nebula框架的定制方式包含3类：

• 修改参数值来实现
• 修改启动参数
• 重写特定的行为方法

修改参数值

Nebula包含了较多参数，请参考 Nebula参数清单 章节内容。

修改启动参数

在Program.cs的Main方法，我们需要执行一个调用

AppStartup.RunAsXXXXXX("xxxxxxx", args, startOption);

第三个参数就是启动参数，它的类型是AppStartOption，可以查看这个类型的成员获知有哪些参数可供调整。

框架行为定制

对于一些特殊场景，Nebula的默认行为可能不满足需求，那么就需要通过扩展的方式来定制框架行为。

下面将介绍如何通过扩展方式来定制Nebula的行为。

开发过程

1. 在VS解决方案树中找到你的项目
2. 找到Customize文件夹，如果不存在则创建
3. 新建一个名为 MyNebulaBehavior 的类名，让它从NebulaBehavior继承
4. 根据你的需求找到并重写NebulaBehavior定义的虚方法
5. 找到Program.cs的Main方法
6. 创建一个AppStartOption的实例，并设置NebulaBehavior属性

示例代码如下：

namespace Nebula.Venus.Customize;

public class VenusNebulaBehavior : NebulaBehavior
{
   // 重写一些你需要的虚方法
}

Program.cs代码如下：

public class Program
{
    public static void Main(string[] args)
    {
        AppStartOption startOption = new AppStartOption {
            NebulaBehavior = new VenusNebulaBehavior()
        };

        AppStartup.RunAsWebsite("Nebula.Venus", args, startOption);
    }
}

NebulaBehavior的可重写方法清单

public virtual bool AllowCors(NHttpContext httpContext)

public virtual void SetMvcOptions(MvcOptions x)

public virtual void SetMvcJsonOptions(MvcNewtonsoftJsonOptions x)

public virtual void ConfigureUrlRouting(WebApplication app)

public virtual async Task ShowHomePage(HttpContext httpContext)

ASP.NET 扩展支持

本文介绍将如何在Nebula中实现与ASP.NET相关的一些扩展场景：

• 添加 MVC Filter
• 调整 MVC UrlRouting
• 引入新组件
• 添加 Middleware/RequestDelegate

添加 MVC Filter

你可以开发一个符合 ASP.NET 要求的 MVC Filter，例如：

public class MyActionFilter : IActionFilter
{
    public void OnActionExecuting(ActionExecutingContext context)
    {
        // 过程省略
    }

然后，在NebulaBehavior的继承类中注册它，例如：

public class MyNebulaBehavior : NebulaBehavior
{
    public override void SetMvcOptions(MvcOptions opt)
    {
        opt.Filters.Add(typeof(MyActionFilter));
    }

最后，在 AppStartOption 中引 MyNebulaBehavior，例如：

public class Program
{
    public static void Main(string[] args)
    {
        AppStartOption startOption = new AppStartOption {
            NebulaBehavior = new MyNebulaBehavior()
        };
        
        AppStartup.RunAsWebsite("XDemo.WebSiteApp", args, startOption);
    }        
}

注意：继承NebulaBehavior的子类必须使用上面的方式来使用它，本文后面将不再重复指出。

调整 MVC UrlRouting

以下代码是 NebulaBehavior 中的 UrlRouting 默认实现：

public virtual void ConfigureUrlRouting(WebApplication app)
{
    app.UseRouting();

    app.MapControllers();

    if( AppStartup.StartOption.EnableDefaultHomePage ) {
        app.MapGet("/", ShowHomePage);
    }
}

说明：

• 建议不要覆盖重写这个方法，可以在调用 base 方法的 before, after 中添加路由规则
• 或者采用 RequestDelegate 的方式来拦截请求实现路由规则

引入新组件

首先要清楚，在 ASP.NET Core 的设计中，一个组件的引入被分为2个阶段：

• 在Ioc容器中注入服务(组件)，对应 Startup.ConfigureServices 方法
• 配置ASP.NET管道（可选步骤），对应 Startup.Configure 方法

在Nebula框架中，不再需要每个项目再维护Startup类型，
只需要开发一个新类型从BaseAppStarter继承，然后选择要实现的方法即可。

BaseAppStarter 提供以下方法用于定制ASPNETCORE的启动过程：

• PreNebulaInit：默认行为：什么也不做(等着你来填空)
• PostNebulaInit：默认行为：什么也不做(等着你来填空)
• PreConfigureServices：默认行为：什么也不做(等着你来填空)
• PostConfigureServices：默认行为：什么也不做(等着你来填空)
• PreConfigureWeb：默认行为：什么也不做(等着你来填空)
• PostConfigureWeb：默认行为：什么也不做(等着你来填空)
• PreBuildWebHost：默认行为：什么也不做(等着你来填空)
• PostBuildWebHost：默认行为：什么也不做(等着你来填空)
• PreApplicationInit：默认行为：什么也不做(等着你来填空)
• PostApplicationInit：默认行为：什么也不做(等着你来填空)

这里以引入 SignalR 为例演示整个开发过程，完整示例如下：

public class ImportSignalrStarter : BaseAppStarter
{
    public override void ConfigureServices(IServiceCollection services)
    {
        string conn = ConfigClient.Instance.GetSetting("Redis.Connection");

        services.AddSignalR(option => {
            option.EnableDetailedErrors = true;
            option.MaximumReceiveMessageSize = 1024 * 1024;
            option.HandshakeTimeout = TimeSpan.FromSeconds(60);
            option.StreamBufferCapacity = 5000;
        }).AddNewtonsoftJsonProtocol()
            .AddStackExchangeRedis(conn);
    }

    public override void PostConfigureWeb(WebApplication app)
    {
        app.UseEndpoints(endpoints => {
            endpoints.MapHub<NotifyHub>("/v20/api/ws/notify", option => {
                option.Transports = HttpTransportType.WebSockets;
            });
        });
    }
}

说明：

• 一个项目中，可以有多个 BaseAppStarter 的继承类
• BaseAppStarter 的继承类必须是 public 的可见性，框架会自动识别并调用。
• BaseAppStarter 的继承类可以出现在类库中，用来构造完整的业务模块（支持所谓的模块化开发）

添加 Middleware/RequestDelegate

开发 MyRequestDelegate，示例代码如下：

public class MyRequestDelegate
{
    private readonly Microsoft.AspNetCore.Http.RequestDelegate _next;

    private long _requestCount = 0;

    public MyRequestDelegate(Microsoft.AspNetCore.Http.RequestDelegate next)
    {
        _next = next;
    }

    public async Task InvokeAsync(Microsoft.AspNetCore.Http.HttpContext httpContext)
    {
        long count = Interlocked.Increment(ref _requestCount);

        httpContext.Response.Headers.Add("x-ReuestCount", count.ToString());

        await _next(httpContext);
    }
}

开发一个新类型，从 BaseAppStarter 继承，

public class XxxAppStarter : BaseAppStarter
{
    public override void PostConfigureWeb(WebApplication app)
    {
        app.UseMiddleware<MyRequestDelegate>();
    }
}

这样就可以了，框架会自动识别！

HTTP管道扩展

在ClownFish框架中，允许采用经典的 HttpModule/HttpHandler 模型来实现HTTP管道的扩展。

开发HttpModule

示例代码如下：

public class QueryModule : NHttpModule
{
    public override void ResolveRequestCache(NHttpContext httpContext)
    {
        // 检查当前请求是否映射到一个已定义的“服务接口”
        ApiAction action = ActionResolver.GetAction(httpContext);
        if( action == null )
            return;


        QueryHandler handler = new QueryHandler(action);
        httpContext.PipelineContext.SetHttpHandler(handler);
    }        
}

小结：

• 开发过程和经典的 ASP.NET HttpModule类似，
• 注意这里是 NHttpModule 基类，
• 然后是重写基类的虚方法（而不是订阅事件）
• 由HttpModule负责HttpHandler的实例化
• 调用httpContext.PipelineContext.SetHttpHandler(handler);来启用HttpHandler实例

开发HttpHandler

（部分）示例代码如下：

internal class QueryHandler : IAsyncNHttpHandler
{
    private NHttpContext _httpContext;
    private ApiAction _action;
    private IDictionary<string, object> _args;


    public QueryHandler(ApiAction action)
    {
        _action = action;
    }

    public async Task ProcessRequestAsync(NHttpContext httpContext)
    {
        _httpContext = httpContext;

        if( await ValidateRequest() == false )
            return;

        _args = GetArgs();

        switch( _action.SourceType ) {
            case 1:
                await new SqlQueryExecutor(_httpContext, _action, _args).ExecuteAsync();
                break;

            case 2:
                await new HttpExecutor(_httpContext, _action, _args).ExecuteAsync();
                break;

            default:
                throw new InvalidOperationException($"无效的 SourceType = {_action.SourceType} 取值。");
        }
    }
}

小结：

• HttpHandler要实现IAsyncNHttpHandler接口
• 然后实现 ProcessRequestAsync 方法

检查 HttpModule 是否已加载

可以访问 Venus 查看进程的 【查看系统信息】，如下下图：

这里显示了所有已加载的HttpModule

框架事件

以下这些框架事件，可用于一些特殊的扩展需求。

HttpClientEvent

public static class HttpClientEvent
{
    /// <summary>
    /// 创建Request对象前将会引发此事件
    /// </summary>
    public static event EventHandler<BeforeCreateRequestEventArgs> OnCreateRequest;
    
    /// <summary>
    /// 创建HttpWebRequest前将会引发此事件，提供最后一个修改请求参数的机会。
    /// </summary>
    public static event EventHandler<BeforeSendEventArgs> OnBeforeSendRequest;

    /// <summary>
    /// 表示请求完成时触发的事件
    /// </summary>
    public static event EventHandler<RequestFinishedEventArgs> OnRequestFinished;

等价的DiagnosticListener接口：

• DiagnosticListener("ClownFish.HttpClientEvent")
  • OnBeforeSendRequest
  • OnRequestFinished

DbContextEvent

namespace ClownFish.Data;

public static class DbContextEvent
{
    /// <summary>
    /// 连接打开事件
    /// </summary>
    public static event EventHandler<OpenConnEventArgs> OnConnectionOpened;

    /// <summary>
    /// 命令执行之前事件
    /// </summary>
    public static event EventHandler<ExecuteCommandEventArgs> OnBeforeExecute;

    /// <summary>
    /// 命令执行之后事件
    /// </summary>
    public static event EventHandler<ExecuteCommandEventArgs> OnAfterExecute;

    /// <summary>
    /// 提交事务事件
    /// </summary>
    public static event EventHandler<CommitTransEventArgs> OnCommited;

等价的DiagnosticListener接口：

• DiagnosticListener("ClownFish.DALEvent")
  • ConnectionOpened
  • BeforeExecute
  • AfterExecute
  • OnCommit

ClownFish.Log

public static class LogHelper
{
    /// <summary>
    /// 写日志时出现异常不能被处理时引用的事件
    /// </summary>
    public static event EventHandler<ExceptionEventArgs> OnError;

public static class FatalErrorLoger
{
    /// <summary>
    /// 当调用 LogException 方法时引发的事件，这表示有一个异常已发生。
    /// </summary>
    public static event EventHandler<ExceptionEventArgs> OnLogException;

RedisClientEvent

public static class RedisClientEvent
{
    /// <summary>
    /// 每当执行一次Redis调用后触发的事件
    /// </summary>
    public static event EventHandler<RedisClientEventArgs> OnExecuteFinished;

RedisClientEvent

public static class RabbitClientEvent
{
    /// <summary>
    /// 每次发送一条消息后触发
    /// </summary>
    public static event EventHandler<SendMessageEventArgs> OnSendMessage;

多数据库支持

ClownFish 提供了【多数据库支持】的特性，直接可支持以下数据库（已适配）：

• SQLSERVER
• MySQL
• PostgreSQL
• SQLite
• 达梦

为了更好的支持MySQL，还同时支持MySql.Data和MySqlConnector 2种不同的客户端，可参考：MySQL异步

开发层面的差异

在ADO.NET基础上，不同的数据库在开发层面主要有以下方面的差异：

• 获取一个标识符的完整形式。拿SQLSERVER来说，传入 xxx 则应该返回 [xxx]
• 判断异常是不是由于 【重复INSERT】 导致的异常
• 获取与数据库类型匹配的SQL语句中的参数名称，一般情况：@name
• 获取与数据库类型匹配的命令集合中参数名称，一般情况：@name
• 根据一条查询语句生成对应的2个分页查询 (List+Count)
• 切换数据库，一般情况：use db1
• 获取新产生的自增列的值

在ClownFish中定义了一个抽象类来隔离不同的数据库差异：

public abstract class BaseClientProvider

为了支持本文开头列出的4种数据库，ClownFish 内置了7个实现类：

• MsSqlClientProvider 支持 SQLSERVER，基于 System.Data.SqlClient
• MsSqlClientProvider2 支持 SQLSERVER，基于 Microsoft.Data.SqlClient
• MySqlDataClientProvider 支持 MySQL，基于 MySql.Data
• MySqlConnectorClientProvider 支持 MySQL，基于 MySqlConnector
• PostgreSqlClientProvider 支持 PostgreSQL，基于 Npgsql
• SQLiteClientProvider 支持 SQLite，基于 System.Data.SQLite
• DaMengClientProvider 支持 达梦，基于 DmProvider

而且还内置了2个"过时的"实现类供参考：

• OdbcClientProvider
• OledbClientProvider

如果遇到不能支持的数据库类型（例如：Oracle）, 请自行实现ClientProvider

特别注意事项
ClownFish虽然支持多种数据库客户端，但ClownFish本身没有引用那些客户端！
全部采用的是反射方式实现。
如果程序启动时出现异常且有XXX程序集找不到时，请在项目中自行引用，然后还要在代码中访问某个类型。

客户端注册方法

可调用以下方法：

/// <summary>
/// DbProviderFactory的辅助工具类
/// </summary>
public static class DbClientFactory
{
    /// <summary>
    /// 注册数据客户端提供者实例
    /// </summary>
    /// <param name="providerName">客户端提供者名称</param>
    /// <param name="provider">提供者实例</param>
    public static void RegisterProvider(string providerName, BaseClientProvider provider)

定制用户身份

默认情况下，ClownFish会在请求进入时识别用户身份，并构造相关一系列身份对象。

对于开发者来说，只需要一个调用就可以获取到用户身份信息：

// 下面代码可在 Controller 代码中使用
IUserInfo userinfo = this.NHttpContext.GetUserInfo();

再或者可以使用下面方式来做授权检查

[Authorize(Roles = "Admin,AppClient")]

在这些代码的背后，ClownFish需要做2件事情：

• 识别用户身份并构造LoginTicket对象，默认来源有2个（名称可配置）：请求头 或者 Cookie
• 根据LoginTicket对象设置 httpContext.User 属性

默认情况下，httpContext.User 属性的值是一个 NbPrincipal 对象。

数据来源扩展

如果你需要使用其它的身份标识来源，例如：

• 从 Redis 中加载已登录的用户身份 （其实是一种Session实现方式）
• 从其它格式的Cookie或者Header中加载用户身份

那么可以参考下面示例。

public sealed class DemoAuthenticateModule : NHttpModule
{
    public override void AuthenticateRequest(NHttpContext httpContext)
    {
        // 从其它来源获取用户身份
        IUserInfo userinfo = GetUserInfo(...);

        // 构造登录身份凭证对象
        LoginTicket ticket = new LoginTicket{ User = userinfo };

        // 设置 httpContext.User
        httpContext.User = new NbPrincipal(ticket, LoginTicketSource.Others);
    }
}

身份类型扩展

ClownFish在内部会使用 NbPrincipal和 NbIdentity 来封装用户身份相关数据。

如果你需要在请求中维护更多信息，也可以自行实现下面2个接口：

• IPrincipal，这个接口是 httpContext.User 的类型要求
• INbIdentity，这个接口是包含用户身份的数据对象

示例代码

public sealed class MyIdentity : ClaimsIdentity, INbIdentity
{
     public IUserInfo UserInfo { get; }

     public Xxxxx OtherData { get; }  // 其它身份数据

     // 这里省略构造方法相关代码
}

public sealed class MyPrincipal : ClaimsPrincipal
{
    // 省略数据成员定义相关代码

    public MyPrincipal(LoginTicket ticket)
    {
        this.AddIdentity(new MyIdentity(ticket.User, ....otherData));
    }
}


public sealed class MyAuthenticateModule : NHttpModule
{
    public override void AuthenticateRequest(NHttpContext httpContext)
    {
        // 从其它来源获取用户身份
        IUserInfo userinfo = GetUserInfo(...);

        // 构造登录身份凭证对象
        LoginTicket ticket = new LoginTicket{ User = userinfo, ....};

        // 设置 httpContext.User
        httpContext.User = new MyPrincipal(ticket);        
    }
}

使用Nebula开发独立应用程序

虽然Nebula开发框架的设计目标是支持 SaaS + 微服务架构，但它也可以开发一些独立的应用程序，例如：

• 独立的小网站
• 独立的WebApi服务

这些独立的应用程序通常有以下特点：

• 只需要 一个应用程序进程 就能满足业务需求，就像小工具一样，运行起来就可以做活了！
• 不依赖配置服务之类的基础服务
• 也不需要多租户的支持

本文将介绍这种应用程序的开发过程。

开发过程

在这个过程，主要是设置一个启动参数（EnableConfigService = false），例如：

public class Program
{
    public static void Main(string[] args)
    {
        AppStartOption startOption = new AppStartOption {

            // 独立运行，不使用配置服务
            EnableConfigService = false,
        };
        
        AppStartup.RunAsWebsite("MyTest.WebSiteApp", args, startOption);
    }        
}

此时，这个项目就可以独立运行了！

参数配置

应用程序的配置参数有以下类别：

• 本地配置参数
  • 环境变量（系统级/用户级/进程级）
  • AppConfig（ClownFish.App.config）
• 远程配置参数
  • 全局配置参数
    • 连接类参数：RabbbitMQ/Redis/es/oss/...
    • 帐号类参数：邮箱帐号，IM应用帐号
    • 密码密钥类：登录密码，JWT密钥
    • URL地址：内部服务，外部服务
  • 数据库连接参数
  • 独立配置文件
  • 远程配置文件（xxx.App.config, xxx.Log.config）

经过前面的步骤，程序是可以独立运行了，但是由于不依赖配置服务，那么"远程配置参数"在哪里配置和读取呢？

答案是：

• 全局配置参数：从 ClownFish.App.config 中获取
• 数据库连接参数：从 ClownFish.App.config 中获取
• 独立配置文件：放在 {app}/_config/ 目录下
• 远程配置文件：此场景下不适用！

例如：

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
    <appSettings>       
        <add key="MySqlClientProviderSupport" value="2" />
        
        <!-- 必要的3个配置参数 -->
        <add key="CLUSTER_ENVIRONMENT" value="BPM-OP"/>
        <add key="Nebula_Environment_Key" value="4fa5e8a12eb04a0b9d8f4330874cfc5a"/>
        <add key="Nebula_Authentication_SecretKey" value="892091a889304639bd42c7ebf61cbf17"/>

        <!-- 一些普通的配置参数 -->
        <add key="key1" value="abcd"/>
        <add key="key2" value="123"/>

        <!-- 连接类参数 -->
        <add key="my_Rabbit" value="server=RabbitHost;username=fishli;password=1qaz7410;vhost=nbtest" />
        <add key="Redis_Connection" value="RedisHost,password=1qazxsw2"/>

        <!-- 非关系型数据库连接（ADO.NET不支持） -->
        <add key="my_Elasticsearch" value="Server=ElasticHost:9200;"/>

        <!-- 帐号类参数 -->
        <add key="mail_config" value="Host=smtp.xxx.com;Port=587;UserName=xxxxx@abc.com;Password=xxx;IsSSL=1"/>
        <add key="WxWork_AppAuth_Config" value="ImType=WxWork;AgentId=999;AppId=xxxxx;AppSecret=xxxx"/>
        <add key="DingDing_AppAuth_Config" value="ImType=DingDing;AgentId=999;AppId=xxxxx;AppSecret=xxxx"/>
        <add key="FeiShu_AppAuth_Config" value="ImType=FeiShu;AgentId=0;AppId=xxxxx;AppSecret=xxxx"/>

        <!-- URL地址 -->
        <add key="UserService_Url" value="http://LinuxTest:8220"/>
     </appSettings>


    <!-- 数据库连接配置 -->
    <connectionStrings>
        <add name="db1" providerName="System.Data.SqlClient"
			 connectionString="server=MsSqlHost;database=MyNorthwind;uid=user1;pwd=xxxx"/>

        <add name="db2" providerName="MySql.Data.MySqlClient"
			 connectionString="server=MySqlHost;database=MyNorthwind;uid=user1;pwd=xxxx"/>

        <add name="db3" providerName="Npgsql"
            connectionString="Host=PgSqlHost;database=mynorthwind;Username=postgres;Password=xxxx"/>
    </connectionStrings>

    <!-- 或者把数据库连接配置在这里 -->
    <dbConfigs>
        <add name="s1" dbType="SQLSERVER" server="MsSqlHost" database="MyNorthwind" uid="user1" pwd="xxxxxx" args="" />
        <add name="s2" dbType="SQLSERVER" server="MsSqlHost" database="MyNorthwind" uid="user1" pwd="xxxxxx" args="" />
        <add name="m1" dbType="MySQL" server="MySqlHost" database="MyNorthwind" uid="user1" pwd="xxxxxx" args="" />
        <add name="m2" dbType="MySQL" server="MySqlHost" database="MyNorthwind" uid="user1" pwd="xxxxxx" args="Allow Zero Datetime=True;Convert Zero Datetime=True;" />
        <add name="pg1" dbType="PostgreSQL" server="PgSqlHost" database="mynorthwind" uid="postgres" pwd="xxxxxx" args="" />
        <add name="dm1" dbType="DaMeng" server="DamengHost" port="5237" database="MyNorthwind" uid="SYSDBA" pwd="xxxxxx" args="" />
    </dbConfigs>
</configuration>

读取参数的示例代码：

string value1 = LocalSettings.GetSetting("key1");

using RabbitClient rabbit = new RabbitClient("my_Rabbit", "default_conn");

IDatabase db = Redis.GetDatabase(2);  // 使用的默认参数名称:Redis.Connection

static readonly NbEsClient esclient = new NbEsClient("my_Elasticsearch");

MailClient client = new MailClient("mail_config");

ImAppMsgClient client = new ImAppMsgClient("WxWork_AppAuth_Config");
ImAppMsgClient client = new ImAppMsgClient("DingDing_AppAuth_Config");
ImAppMsgClient client = new ImAppMsgClient("FeiShu_AppAuth_Config");


using DbContext db1 = DbContext.Create("db1");
using DbContext db2 = DbContext.Create("db2");
using DbContext db3 = DbContext.Create("db3");

using DbContext s1 = DbContext.Create("s1");
using DbContext m1 = DbContext.Create("m1");

不启用配置服务的影响

当设置 EnableConfigService = false 之后，会带来以下影响：

• 不再访问配置服务
  • 默认只能读取本地配置参数
• Venus无法发现当前进程
  • 不统计运行状态，以及健康状态识别
  • 无法查看内部状态
  • 不支持指标监控
  • 不支持告警通知
• 禁用 业务指标监控
• 禁用 可用性监控
• 禁用 跨进程的事件通知

使用ClownFish开发独立Web应用程序

虽然可以 使用 Nebula 开发独立应用程序，但是 直接使用 ClownFish开发独立应用程序 会有以下优点：

• 发布目录更干净，因为ClownFish的第三方依赖更少
• 可定制性更强

ClownFish和Nebula的设计理念不一样：

• Nebula：All in One，大而全，开箱即用。
• ClownFish：模块化，小积木，自行组装。

ClownFish 有以下几个 Nuget 包，可根据需求自行引用：

• ClownFish.net：最基本单元，包含开发单个应用程序的所有最基本功能。
• ClownFish.Web：基于ClownFish.net，支持 Asp.net Core
• ClownFish.Redis：基于ClownFish.net的Redis工具类库
• ClownFish.Rabbit：基于ClownFish.net的RabbitMQ工具类库
• ClownFish.ImClients：基于ClownFish.net的IM客户端工具类库，支持：企业微信，钉钉，飞书
• ClownFish.Email：基于ClownFish.net的Email简单封装工具类库
• ClownFish.Office：对个别典型场景的Office文档操作做了一点简单的封装

1，创建新项目

打开 Visual Studio (最新版本) 创建一个 你需要的 应用程序…………

2，添加包引用

例如：

<ItemGroup>
    <PackageReference Include="ClownFish.Web" Version="9.25.1105.1" />
    <PackageReference Include="MySqlConnector" Version="2.2.7" />
</ItemGroup>

说明：

• ClownFish.Web：它包含了 ClownFish.net
• MySqlConnector：因为程序需要访问 MySQL 数据库，如果还需要访问其它类别数据库，请自行引用

3，修改 Program.cs

一个典型的WebApi项目的启动代码可参考：

public class Program
{
    public static void Main(string[] args)
    {
        AspnetCoreStarter.Run(new MyStartup());
    }
}

// 这个类用于定制启动过程
public class MyStartup : WebApplicationStartup
{
    public override bool AutoInitDAL => true;   // 初始化数据访问层

    public override bool AutoInitTracing => true;   // 开启日志与监控

    public override bool AutoInitAuth => true;  // 初始化身份认证模块

    public override void AppInit()
    {
        // 你的初始化代码写在这里…………
    }

    public override void ConfigureServices(IServiceCollection services)
    {
        // 加载所有 Controller
        services.AddControllers(this.RegisterInnerMvcFilters);

        base.ConfigureServices(services);
    }

    public override void ConfigureWeb(WebApplication app)
    {
        app.MapControllers();

        app.MapGet("/", ClownFish.Web.Utils.HttpContextUtils.ShowHomePage);
    }
}

现在这个项目就可以开发ClownFish的三种 代码执行主体

使用ClownFish开发独立Console应用程序

本文介绍直接使用ClownFish来开发一个控制台应用程序。

1，创建新项目

打开 Visual Studio (最新版本) 创建一个 你需要的 应用程序…………

2，添加包引用

例如：

<ItemGroup>
    <PackageReference Include="ClownFish.net" Version="9.25.1105.1" />
    <PackageReference Include="MySqlConnector" Version="2.2.7" />
</ItemGroup>

说明：

• MySqlConnector：因为程序需要访问 MySQL 数据库，如果还需要访问其它类别数据库，请自行引用

3，修改 Program.cs

一个典型的WebApi项目的启动代码可参考：

public class Program
{
    public static void Main(string[] args)
    {
        ConsoleAppStarter.Run(new MyStartup());
    }
}

// 这个类用于定制启动过程
public class MyStartup : ConsoleAppStartup
{
    public override bool AutoInitDAL => true;   // 初始化数据访问层

    public override bool AutoInitTracing => true;   // 开启日志与监控

  
    public override void AppInit()
    {
        // 你的初始化代码写在这里…………
    }

}

现在这个项目就可以开发ClownFish的2种 代码执行主体

开发模块化的应用程序

Nebula支持多种开发模式：

• 微服务架构
• 单体应用程序
• 模块化的-应用程序

本文将主要介绍第3类：模块化的-应用程序

模块化开发

对于复杂的业务范围，我们可以将其拆分为多个业务单元（模块），例如：

• 用户模块
• 订单模块
• 商品模块
• 仓库模块
• 运营模块

我们可以：

• 在一个源代码项目中实现所有的这些业务需求，即：单体应用程序
• 也可以将它们拆分成多个源代码项目，每个业务模块对应一个项目，即：模块化应用程序

根据 代码执行主体 中的介绍，几乎所有业务需求最后会以3类技术方式实现：

以上3类实现方式，ClownFish将它们抽象为3类代码执行主体：

• HttpAction (由ASP.NET支持)
• MessageHandler
• BackgroundTask

这3类代码执行主体可以出现在任何类型的应用程序项目中，因此，我们可以按业务模块来组织源代码项目：

• 每个业务模块对应一个源代码项目
• 每个源代码项目中，根据业务需求来实现 Controller, MessageHandler, BackgroundTask
• 然后在一个HOST项目中引用这些（或者全部）源代码项目（或者nuget包）

最终部署目标

一套代码，满足2种部署需求：

• 单体应用部署
  • 只部署一个应用（包含所有业务功能）
  • 通常用于 私有化部署
• 微服务部署
  • 一个模块对应一个微服务，所有微服务的共同支撑整个业务功能
  • 通常用于 云上部署

说明：

• 每个业务模块可采用微服务架构开发，对应一个Web项目，可独立运行
• AllInOneWeb项目，本身不需要实现任何业务逻辑，只需要引用其它业务模块项目引可

模块化开发过程

模块化开发过程中可能会遇到的问题是：

• 个别模块的源代码项目有特殊的初始化需求，例如用户模块需要注册 IdentityServer

这时就需要执行 ASP.NET 的 ConfigureServices 方法，解决方法可参考下面代码示例。

只需要2个步骤：

• 在程序集中标记 [assembly: ClownFish.Base.ApplicationPartAssembly]
• 在源代码项目中实现一个从BaseAppStarter继承的 公开 类型，例如：

public class DemoAppStarter : BaseAppStarter
{
    public override void PostConfigureServices(IServiceCollection services)
    {
        Console2.Info("DemoAppStarter.PostConfigureServices -----------");
    }

    public override void PostConfigureWeb(WebApplication app)
    {
        Console2.Info("DemoAppStarter.PostConfigureWeb -----------");
    }

    // 还有一些虚方法就不一一列举了
}

支持SaaS私有化部署

由于种种原因，SaaS应用程序就是需要在客户环境中私有化部署！！

那么如何让一套代码既支持云上的标准SaaS模式，又支持客户的私有部署模式呢？

本文将介绍如何这现这个需求。

首先请阅读以下二篇文档

• SaaS多租户架构--数据库
• SaaS多租户架构--应用程序

准备配置参数

为了保证一套代码能同时支持SaaS和OP二种部署模式，且部署过程简单，请先设计好2个配置参数：

• Default_Tenant_Id=xxxxxxxxxxxxxxxxxxx：默认租户ID
• Default_Tenant_ConnName=tenant_db：默认的租户库连接名称

用户登录

租户的切换其实是根据用户登录身份来实现的，具体可参考IUserInfo的定义

public interface IUserInfo
{
    string TenantId { get; }
    string UserId { get; }

因此，可以按以下方式来实现：

• 为这个客户分配一个明确的 【租户ID】，例如：TenantId=fc6e4bc7a3484304be1afb65b999938d
• 私有部署环境中，给程序添加环境变量

docker run .....
  -e Default_Tenant_Id=fc6e4bc7a3484304be1afb65b999938d \
  ........

• 定义静态变量保存这个默认租户ID

public static readonly string DefaultTenantId = LocalSetting.GetSetting("Default_Tenant_Id");

• 应用程序的显示登录表单时，如果SaasOpTenantId有值，就不显示 "租户CODE" 这种输入控件
• 用户登录时，设置 UserInfo.TenantId

WebUserInfo userinfo = new WebUserInfo{
    TenantId = DefaultTenantId // 从静态变量中取值
    UserId = "U0001",
    UserName = "张三",
    UserRole = "NormalUser"
}
AuthenticationManager.Login(userInfo, seconds);

Controller中获取租户Id

string tenantId = this.GetTenantId();

数据库操作

这块的代码实现与标准的SaaS没有任何区别，也是：

// Controller
using DbContext dbContext = this.CreateMasterConnection();
using DbContext dbContext = this.CreateAppDbConnection("logging");
using DbContext dbContext = this.CreateTenantConnection(tenantId);
using DbContext dbContext = this.CreateTenantConnection();

// MessageHandler, BackgroundTask
using DbContext dbContext = DbConnManager.CreateMaster();
using DbContext dbContext = DbConnManager.CreateAppDb("logging");
using DbContext dbContext = DbConnManager.CreateTenant(tenantId);

说明：

• 如果在程序启动时设置了 EnableConfigService=false，租户库的连接名称将是 Default_Tenant_ConnName 参数指定的名称，不再按SaaS的规则去做映射查找。
• 主库的连接名称依然是 master
• 应用库根据使用情况来配置

数据库连接配置如下：

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
    <appSettings>        

     </appSettings>

    <connectionStrings>
        <add name="master" providerName="System.Data.SqlClient"
			 connectionString="server=MsSqlHost;database=master;uid=user1;pwd=xxxx"/>

        <add name="logging" providerName="MySql.Data.MySqlClient"
			 connectionString="server=MySqlHost;database=logging;uid=user1;pwd=xxxx"/>

        <add name="tenant_db" providerName="System.Data.SqlClient"
			 connectionString="server=MsSqlHost;database=tenant_db;uid=user1;pwd=xxxx"/>

    </connectionStrings>
</configuration>

以【单体应用】方式运行

如果私有化部署时希望采用单体应用的方式运行，可以参考下面二篇文档：

• 独立应用程序
• 模块化开发

最后可以实现的效果是：

• 一套代码支持 SaaS 和 OP 二种部署模式
• 一套代码支持微服务开发架构 和 单体应用程序架构

部署过程

与云端的部署方式一样，部署所有服务及数据库。

唯独：租户总表只有一条记录。

使用其它的配置服务

Nebula是一个基于 .NET 技术的开发框架体系，核心特色是【支持SaaS多租户】、【支持微服务开发】和【实时可观测能力】。

这是Nebula的设计目标。

为了实现这些能力，Nebula的许多功能都依赖于 自带的 配置服务，例如下面这些API：

• ConfigClient.Instance.GetSetting(string name)
• Settings.GetSetting(string name)
• ConfigFile.GetFile(string name)
• DbConnManager.CreateMaster()
• DbConnManager.CreateAppDb(string name)
• DbConnManager.CreateTenant(string tenantId)
• 各种Client：RabbitMQ/Redis/InfluxDB/Mail/IM/.....

考虑到以下二种场景：

• 独立应用程序
• SaaS私有化部署

它们可能并不会启用Nebula自带的 配置服务，甚至使用第三方的配置服务，例如：Nacos

本文将介绍如何解决这个问题：使用其它的配置服务之后，以上API仍能正常工作。

理解配置服务

配置服务只有一个功能：统一管理的各种远程配置参数：

• 全局配置参数
  • 读取方式：Settings.GetSetting(string name)
• 数据库连接参数
  • 读取方式：DbConnManager.CreateAppDb(string name)
• 远程独立配置文件
  • 读取方式：ConfigFile.GetFile(string name)
• 远程配置文件（xxx.App.config, xxx.Log.config）
  • 读取方式：Nebula内部调用

为了实现配置服务的替换，必须将以上参数转成 第三方配置服务能接受的方式：

• 所有 全局配置参数：变成一个配置文件，即 key=value 的数组形式，再做个JSON/XML 序列化即可。
• 所有 数据库连接参数：同上处理
• 远程独立配置文件：保持不变
• 远程配置文件：保持不变

然后将这些文件内容存储在 第三方配置服务。

实现思路

ConfigClient是所有访问配置服务的入口类，只要把它的行为替换就可以了。

Nebula将各配置种读取操作抽象了一个接口：IConfigClient

为了能替代Nebula自带的配置服务，我们有2种方法

• 使用内置的工具类：MemoryConfig，因为内存中的数据总是优先查找的。
• 实现IConfigClient接口，然后调用 ConfigClient.Instance.SetClient 来注册它

1，实现IMoonClient接口，这是一种最更灵活通用的方案。

• 优点：允许你来决定所有的替换细节
• 缺点：需要时间开发

2，MemoryConfig 是一种简单且内置的实现，可以理解为：

• 将程序所需要的配置参数直接放在内存中
• 当需要访问配置服务时，Nebula会自动切换到内存中获取它们

它提供以下方法来接收数据：

public static void AddSetting(string name, string value)
public static void AddDbConfig(string name, DbConfig config)
public static void AddFile(string name, string fileText)
public static void SetAppConfig(string xml)
public static void SetLogConfig(string xml)

本文将介绍MemoryConfig这种方式来实现。

加载配置参数

有了前面的步骤，所有配置参数都存储在第三方配置服务中，程序就可以读取它们。

如何从第三方配置服务中读取参数，请参考相关文档，此处忽略！

在程序启动时，先将本程序所需的各种参数读取到，然后调用MemoryConfig相关方法即可，例如：

public override void PreNebulaInit()
{
    NacosConfig config = LocalSetting.GetSetting<NacosConfig>("NacosConfig");
    NacosClient client = new NacosClient(config);

    // 假设用应用程序的名称做命名空间
    string @namespace = AppStartup.RuntimeStatus.ApplicationName;


    // 加载 全局配置参数
    string settings = client.GetValue(@namespace, "settings.json");
    settings.FromJson<List<NameValue>>().ForEach(x => MemoryConfig.AddSetting(x.Name, x.Value));

    // 加载 数据库连接参数
    string conns = client.GetValue(@namespace, "db_conns.json");
    conns.FromJson<List<DbConfig>>().ForEach(x => MemoryConfig.AddDbConfig(x.Name, x));

    // 加载 远程独立配置文件
    string file1 = client.GetValue(@namespace, "file1.xml");
    MemoryConfig.AddFile("file1", file1);

    // 加载 远程配置文件
    string appConfig = client.GetValue(@namespace, "App.Config");
    MemoryConfig.SetAppConfig(appConfig);
}

包装Action返回结果

许多开发人员喜欢给Action的返回结果再做一层包装，例如：

{
  "code": 1,
  "message": "xxxxxxxx",
  "data": {这里是原本Action的返回结果}
}

效果演示

Nebula对这种需求提供了内置支持，不用考虑这个包装对象的存在，Action只返回该返回的数据就可以了，例如：

[Route("return/namevalue")]
public NameValue ReturnObject2()
{
    return new NameValue("abc", "12345");
}

服务端的返回结果：

HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8

{
    "code": 1,
    "message": "",
    "detailMessage": "",
    "data": {
        "Name": "abc",
        "Value": "12345"
    }
}

如果遇到异常

[Route("return/error")]
public int ReturnError()
{
    if( DateTime.Now.Year > 2000 )
        throw new MessageException("一个测试消息异常") { StatusCode = 555 };
    else
        return 111;
}

服务端的返回结果：

HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8

{
    "code": 555,
    "message": "一个测试消息异常",
    "detailMessage": "ex.ToString()",
    "data": ""
}

如何实现？

1，开启一个开头（默认是关闭状态）

在ClownFish.App.config中添加：

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
    <appSettings>
        <add key="Nebula_WrapDataResultFilter_Enable" value="1" />
    </appSettings>    
</configuration>

或者用环境变量也可以，例如：

docker run -d --name xxxxx  \
  -e Nebula_WrapDataResultFilter_Enable=1 \

2，对返回结果做打包：

先定义包装类型：

public record WrapData(int code, string message, string detailMessage, object data);

再重写 NebulaBehavior.WrapActionResult 方法。

public override object WrapActionResult(object resultOrException)
{
    // Action可能会返回NULL
    if( resultOrException == null) {
        return new WrapData(1, "", "", "");
    }

    // Action可能返回了一个简单的：数字/字符串
    if( resultOrException is string || resultOrException.GetType().IsPrimitive ) {
        return new WrapData(1, "", "", resultOrException.ToString());
    }

    // Action中出现了异常
    if(  resultOrException is Exception ex ) {
        return new WrapData(ex.GetErrorCode(), 
                    ex.Message,
                    "ex.ToString()",   // 实际使用时，不需要用双引号
                    "");
    }

    // Action的“正常”返回数据
    return new WrapData(1, "", "", resultOrException);
}

3，调用API

GET http://localhost:8206/v20/api/WebSiteApp/wrapdata/return/namevalue HTTP/1.1
x-wrap-result: 1

注意上面的调用中，有一个 x-wrap-result 的请求头。

如果没有这个请求头，结果将不会做包装！

谈谈这种API的设计方式

"对Action返回结果再做包装"，最初的设计者可能认为（后来人应该是无脑抄）：

这种API对调用者会比较友好，因为可以不用考虑异常情况，调用代码会比较简单。

他想像中的调用代码：

var result = call_httpapi(...);
if( errorcode == 1){
   // 调用成功
   // 继续执行其它业务操作
}
else{
   // 调用失败
   // ################ 在这里处理异常 111111111111111
}

国内大厂的许多外部API也确实都采用了这种设计方式，但这种做法其实是 非常差劲的！

因为有些异常情况是不受业务代码控制的，例如：

• 网络不通
• 服务不可用
• 服务端出现不可控异常，例如：OOM

在这些场景下，前面所说的包装方式就无效了，最后调用代码还是必须要增加 try...catch 来捕获异常，例如：

try{
  var result = call_httpapi(...);
  if( errorcode == 1){
    // 调用成功
    // 继续执行其它业务操作
  }
  else{
    // 调用失败
    // ################ 在这里处理异常 111111111111111
  }
}
catch(Exception ex){
    // ################ 这里处理其它类型的异常 222222222222222
}

如果业务项目有全局异常处理机制，errorcode != 1 的场景反而会更麻烦，它需要主动抛出一个异常，来结束当前代码块！

综上所述，这种做法虽然在国内很流行，但是设计想法即是非常愚蠢的！
调用代码不仅得不到简化，反而会更复杂，除非调用方意识不到要做异常处理！

响应退出事件/应用程序优雅退出

基于 ClownFish/Nebula 的应用程序，都会响应 Ctrl+C 事件来退出应用程序。

响应 Ctrl+C 事件需要2方面来共同完成：

• 框架层面，包含基础组件，例如：HttpClient, MessageSubscriber, BackgroundTask
• 业务代码

目前框架层面已全面支持，本文主要介绍在业务代码中如何 响应 Ctrl+C 事件

调用阻塞等待API

在.NET的框架设计中，绝大多数的阻塞等待API都可以授受一个CancellationToken参数， 它可以 （在程序退出时）提前结束等待。

因此我们可以在调用阻塞方法时，使用这个参数，可供传递的对象有2个（选择其一）：

• ClownFishInit.AppExitToken ，由ClownFish提供
• AppStartup.AppExitToken ，由Nebula提供

使用示例：

await Task.Delay(waitSeconds * 1000, AppStartup.AppExitToken);
// or
await _channel.Reader.ReadAsync(AppStartup.AppExitToken);

和超时时间一起使用：

using CancellationTokenSource timeoutCancelSource = new CancellationTokenSource(timeout);
using CancellationTokenSource tokenSource = CancellationTokenSource.CreateLinkedTokenSource(timeoutCancelSource.Token, AppStartup.AppExitToken);

var data = await item.Channel.Reader.ReadAsync(tokenSource.Token);

主动判断退出事件

在一个 大循环 中，我们也可以主动去判断有没有触发退出事件，进而提前结束。

示例代码：

if( AppStartup.AppExitToken.IsCancellationRequested ) {
    Console2.WriteLine("Application exit, VenusCommandClient exit.");
    break;
}

注意事项

以下这些设计不会及时响应退出事件

• Thread.Sleep( 一个长时间 )
• for/while 中长时间运行，且不检查是否已发生退出事件

如果程序不能及时响应退出事件，那么在超过一段时间后，会被强制中止线程，最后有可能会出现数据损坏。

2个默认的超时时间：

• asp.net: 30s
• docker: 10s

检验是否正确响应退出事件

• 启动程序
• 稍许等待，让程序所有功能都正常运行 ......
• 按下 Ctrl+C

观察程序是否立即结束，如果不能，则需要结合具体代码添加埋点处理。

请点击目录中的节点查看具体文档

日志组件

本文主要介绍ClownFish内置日志，日志组件，及日志配置相关内容。

执行主体

以下3类场景被ClownFish视做执行主体，为自动为它们产生OprLog日志

• HTTP响应 - HttpAction (Controller)
• 消息处理 - MessageHandler
• 后台任务 - BackgroundTask

内置日志

ClownFish/Nebula内置5类日志

1. 操作日志：每个【执行主体】每次执行都会产生一条操作日志OprLog
2. 异常日志：【执行主体】在发生异常时出现，异常信息会合并到当次OprLog中
3. 性能日志：【执行主体】都会有性能监控，如果超过性能阀值，将产生性能日志，合并到OprLog中
4. 框架性能日志：记录ASP.NET执行过程中耗费的性能，如果超过性能阀值，将产生性能日志 NebulaLog
5. 致命异常日志：会记录导致程序异常退出的异常，错误消息固定写入到系统临时目录下，文件名: {sys-temp-directory}_ClownFish_net_error_{date}.txt

日志组件

日志组件由ClownFish提供，主要特性有：

• 支持结构化日志（非随意性文本）
• 区分日志数据类型和写入器概念
  • 日志数据类型就是一个.NET类型，包含日志的数据成员定义
  • 写入器是指将日志数据对象写入到哪里（实现持久化）
• 一个日志类型可以对应多个写入器（同时写到多个地方）
• 日志采用异步设计，不阻塞调用线程

写日志

可以在代码中调用以下API

/// <summary>
/// 日志记录的工具类
/// </summary>
public static class LogHelper
{
    /// <summary>
    /// 记录指定的日志信息
    /// 说明：此方法是一个异步版本，内部维护一个缓冲队列，每 XXms 执行一次写入动作
    /// </summary>
    /// <typeparam name="T">日志信息的类型参数</typeparam>
    /// <param name="info">要写入的日志信息</param>
    public static void Write<T>(T info) where T: class, ITime

    /// <summary>
    /// 直接将一个异常对象写入到异常日志
    /// </summary>
    /// <param name="ex"></param>
    /// <param name="addition"></param>
    public static void Write(Exception ex, string addition = null)
}

注意事项

虽然LogHelper提供了一个记录异常对象的方法，但是不建议在以下场景中调用

• HTTP请求过程中（Controller，BLL）
• 消息处理过程中（MessageHandler）
• 后台任务中（BackgroundTask）

因为这些执行环境中ClownFish有异常处理机制，会统一记录异常，而且异常日志会有上下文关联信息。

日志配置

在每个项目中，日志的配置文件是ClownFish.Log.config，以下是Nebula类库中的默认配置

<?xml version="1.0" encoding="utf-8"?>
<LogConfig>
    <!--是否启用日志写入功能，总开关-->
    <Enable>true</Enable>
    <!--日志从缓存区写入持久化存储的间隔时间，单位：毫秒-->
    <TimerPeriod>500</TimerPeriod>
    <!--性能日志的阀值参数，超过这个阀值将会认为是“慢”操作，此时会标记 IsSlow=1，并记录请求，单位：毫秒-->
    <Performance HttpExecute="100" HandleMessage="200"  />
    <!-- 【本地】日志文件相关配置 -->
    <File RootPath="Logs" MaxLength="500M" MaxCount="100" />

    <Writers>
        <!--定义几个内置的写入器 -->
        <Writer Name="Xml" Type="ClownFish.Log.Writers.XmlWriter, ClownFish.net" />
        <Writer Name="Json" Type="ClownFish.Log.Writers.JsonWriter, ClownFish.net" />
        <Writer Name="Json2" Type="ClownFish.Log.Writers.Json2Writer, ClownFish.net" />
        <Writer Name="Http"  Type="ClownFish.Log.Writers.HttpJsonWriter, ClownFish.net" />
        <Writer Name="ES"    Type="ClownFish.Log.Writers.ElasticsearchWriter, ClownFish.net" />
        <Writer Name="Rabbit2" Type="ClownFish.Log.Writers.RabbitHttpWriter, ClownFish.net" />
        <Writer Name="Rabbit" Type="ClownFish.Rabbit.Log.RabbitWriter, ClownFish.Rabbit" />
        
        <!--如果项目中有自定义的写入器，请自行在项目中创建 ClownFish.Log.config 并添加 【你新增的】写入器-->
    </Writers>
    
    <Types>
        <!--定义支持的数据类型，并为每种数据类型配置使用哪种写入器，
            如果某个数据类型同时指定了多个写入器，将会以不同的持久化方式同时记录多次。-->
        <Type DataType="ClownFish.Log.Logging.OprLog, ClownFish.net" Writers="NULL" />
        <Type DataType="ClownFish.Log.Logging.InvokeLog, ClownFish.net" Writers="NULL" />
        <Type DataType="Nebula.Log.NebulaLog, Nebula.net" Writers="NULL" />

        <!--### 注意：1，为能让配置文件通用，这里并没有为每个日志数据类型指定关联的写入器，
                      2，写入器的关联设置在配置参数 ClownFish_Log_WritersMap 中一次性指定  -->
    </Types>
</LogConfig>

以上默认配置中，所有日志数据类型对应的写入器是NULL，即不执行任何写入。

实际使用时，可以在配置服务中增加一个参数项来全局指定：

name: ClownFish_Log_WritersMap
value: InvokeLog=Rabbit;OprLog=es;NebulaLog=es;*=NULL

value的含意是：

• InvokeLog=Rabbit    表示 ClownFish.Log.Logging.InvokeLog 类型将使用 Rabbit 写入器
• OprLog=es;NebulaLog=es    表示 OprLog和NebulaLog，这2类日志使用 ES写入器
• *=NULL     *星号表示其它（未列出的）日志类型，它们将使用 NULL写入器（即丢弃）

InvokeLog=Rabbit;OprLog=es,Json;*=NULL

日志存储与 Elasticsearch 配置

配置日志组件

可以通过配置的方式将所有日志写入ES，这部分请参考：日志组件

最简单的设置就是：

name: ClownFish_Log_WritersMap
value: InvokeLog=Rabbit;*=es

日志存储策略及索引模板

请确保 Elasticsearch 和 Kibana 已正确部署，处于运行状态，
此时可启动 Venus，它会在自动创建所需的 日志存储策略，索引模板，和索引模式。

在Kibana中可以看到

如果这时运行程序，就能产生日志了，最后可以看到这个样子：

Kibana索引模式

为了能在Kibana中查看日志，需要新增索引模式，例如：

查看日志数据

现在可以打开Discover查看日志数据了：

这是比较原始的方式，我们可按下图来定制显示方式

下图是定制后的显示方式:

异常日志

ClownFish会在以下场景中自动记录异常日志

• 处理HTTP请求过程中（HttpAction）
• 处理消息过程中（MessageHandler）
• 后台任务中（BackgroundTask）

最后可以在日志中找到它们，假设我们将日志已写入Elasticsearch

从Venus跳转到日志列表

查看当天的异常日志，可以直接从Venus中跳转，例如：

点击每个服务卡片中的“异常数”（下面的数字）即可跳转到Kibana的具体页面中。

展开后可查看日志详情，其中就包含了详细的异常描述以及完整的请求记录。

致命异常日志

如果程序在启动时出现异常，那么这类异常会导致程序退出，此时就会记录一条【致命异常日志】

这类日志固定写入到系统临时目录下，文件名: {sys-temp-directory}_ClownFish_net_error_{date}.txt

如下图：

而且还会有企业微信通知：

异常处理

ClownFish会在以下场景中自动记录异常日志

• 处理HTTP请求过程中（HttpAction）
• 处理消息过程中（MessageHandler）
• 后台任务中（BackgroundTask）

所以，绝大多数情况下是不需要捕获异常的。

将异常输出到控制台

在某些场景下如果需要小范围的进行异常处理，尤其是涉及 将异常输出到控制台时 ，
那么建议使用下面的做法:

try{
    //.....
}
catch(Exception ex){
    Console2.Error(ex);
}

这样做的好处是可以在Venus中看到异常的累计次数，也可以增加告警规则，例如：

当次数超过一个阀值时，将会以红底色显示。

我们还可以定义一个告警规则，例如：

<Monitor AppName="XDemo.TestApp" MetricName="ERR" 
        Statistic="Incremental" DueTime="40" 
        Period="60" Threshold="100" Compare="MoreThan"/>

规则的含意是：

• 针对 "XDemo.TestApp" 这个应用程序，
• 在程序启动40秒后可始监控，
• 每60秒获取 ERR 这个指标值（异常次数）
• 如果增量超过 100，即产生一条告警通知。

不推荐的做法

try {
    //.............
} catch ( Exception e ) {
    Console2.Info($"刷新告警规则缓存失败 3分钟后重试 失败原因{e.Message}");
}

catch ( Exception ex ) {
    Console2.Info($"队列异常：{ex.Message}-{ex.StackTrace}");
}

catch ( Exception ex ) {
    Console2.Info($"Alert：{ex.Message}-{ex.StackTrace}");
}

catch (Exception ex)
{
    Console.WriteLine(ex.Message + ex.StackTrace);
}

catch (Exception iex)
{
    Console2.Warnning(iex.ToString());
}

catch (Exception ex)
{
    Console.WriteLine(ex);
}

catch (Exception e)
{
    Console.WriteLine($"发送打开消息到rabbitmq异常{e.Message}");
}

catch (Exception e)
{
    Console.WriteLine("快照获取失败" + e.ToJson());
    return null;
}

性能日志

ClownFish会在以下场景中自动监控调用性能并记录日志

• 处理HTTP请求过程中（HttpAction）
• 处理消息过程中（MessageHandler）
• 后台任务中（BackgroundTask）

性能监控范围主要涵盖以下范围：

• 基于HTTP协议的远程调用
  • 微服务之间调用
  • ElasticSearch调用
  • InfluxDB调用
  • VictoriaMetrics调用
  • 3种IM调用（企业微信，钉钉，飞书）
  • 阿里云日志服务调用
  • 阿里云OSS调用
  • 阿里云短信
• 基于TCP协议的远程调用
  • SQL数据库调用（各种RMDBS)
  • Redis调用
  • 发送RabbitMQ消息
  • 发送Plusar消息
  • 发送邮件
• 非 远程调用
  • 读请求流（Request Body）
  • 用户代码（HttpAction）执行前的框架耗时

与APM的比较

ClownFish/Nebula的性能监控日志与其它APM性能监控日志对比，主要有以下优势：

• 覆盖面更完整（3种开发模型）
• 日志量精准：仅当任务的执行时间超过性能阀值时才会记录（不是全量，也不是采样）
  • 日志量较小，不多不少，刚刚好（只要发生一次慢任务，一定会记录）
  • 性能损耗较小：充分利于框架特性 + 延迟技术 + 日志量小
• 可监控普通方法（非远程调用）
• 内容更全，例如：可监控【读请求流】，可包含租户用户等等内部数据
• 整合了最优秀的日志查看工具：Kibana, JeagerUI

利用性能日志排查问题

可参考以下链接：

• 性能问题排查-通用方法
• 性能问题排查-案例1
• 性能问题排查-案例2
• 性能问题排查-案例3

查看调用链路

当我们从日志列表界面Kibana进入日志详情页面JeagerUI时，

此时是按【当前日志记录】为 顶层节点 来展示的。

在微服务架构中，服务之间相互调用可能很常见。很有可能这个【当前日志记录】只是 整个调用链路 中的一个环节而已， 如果此时需要查看 完整的调用链路，可以在 URL 后面添加 ?tree 来实现，例如：

监控普通方法调用

这里所说的【普通方法】是指没有远程调用的方法，全是一些内存操作，也可以称为【CPU密集型操作】。

要监控这些方法的执行性能就是给这些方法增加 [MonitorPerformance] 标记，例如：

这个HttpAction调用了5个 “普通方法”，它们的定义如下：

最后可以在 JeagerUI 看到如下效果：

框架性能日志

Nebula会针对HTTP请求处理生成另外一份日志：框架性能日志

这里所说的框架包含3部分：

• ASP.NET Core
• Nebula
• ClownFish

日志产生条件：框架代码执行时间大于 性能阀值

性能阀值：

• 参数名：Nebula_Performance_ThresholdMs
• 默认值：0
• 数量单位：毫秒
• 配置方式：环境变量，配置服务，本地配置文件
• ES中的索引名称：nebulalog-yyyyMMdd
• 说明：如果参数值小于等于零，则表示不启用。

日志样例：

说明：

• 此处的 OprId 和 OprLog中的OprId相同，目的就是为了方便相互查找。
  比如：在浏览 NebulaLog时，可以根据某个OprId找到对应的HTTP操作日志OprLog，反之也成立。
• UserCode是相对框架代码来说的，一般是指Action的代码
• PreFindAction和PostFindAction之间包含了所有Middleware和Filter在Action【之前】的执行时间
• PostRequestExecute和UpdateRequestCache之间包含了所有Middleware和Filter在Action【之后】的执行时间
• 默认情况下Nebula并没有引入其它Middleware和Filter，如果上面2个阶段产生了耗时，那么都是ASP.NET本身的消耗

ES配置说明

为了能看到上面的截图，需要在ES中配置以下地方：

• 创建一个名为 nebulalog 的 index template，并引入合适的生命周期策略。
• 创建一个名为 nebulalog-* 的 Index pattern

HTTP日志

HTTP日志分为2大块：

• 当次HTTP请求的 Request/Response 日志
• 当次HTTP请求执行过程中，由 HttpClient 发起的HTTP请求及响应的日志

注意：记录日志都是需要耗费性能以及增加内存占用的，
所以 ClownFish 设计了8个参数来控制日志记录哪些内容

重要说明：

• 以上8个参数，可分为4组
  • LogXxx: 是否记录 Request 或者 Request 的 “开始行”和“请求头/响应头”
  • LogXxxBody: 是否记录 请求体 或者 响应体
  • 当 LogXxx = 0 时，会忽略 LogXxxBody 的设置。
• 如果当前HTTP请求执行时间超出性能阀值，则 强制记录请求，
  • 可认为此时(当前请求) ClownFish_Log_Http_LogRequest=1
• HttpClient相关日志仅在 当前HTTP请求执行时间超出性能阀值时有效
  • 此处没有参数可控制，因为没有性能问题的执行过程是不记录性能日志的
  • 如果确实需要记录执行过程，可修改 性能阀值
• 记录 “请求体或者响应体” 有许多限制，例如：
  • 必须有 Body
  • Content-Type 必须为常见的文本类型
  • Body 不能过大，必须小于 RequestBodyBufferSize 的设置
  • Body stream 可重复读取

Body stream 可重复读取

.NET 出于性能以及内存占用的原因，默认的行为是不允许 多次 读取 Request.Body
请求体的第一次读取通常发生在Action方法的参数赋值的时候，
请求执行过程中，就不能再次读取了，因此无法在日志中记录请求体内容，

如果需要让ClownFish在日志中记录请求体内容，则需要按以下步骤来操作。

开启请求缓冲区

可以在 ClownFish.App.config 或者环境变量在设置参数 ClownFish_Aspnet_RequestBufferSize

例如：

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
    <appSettings>
        <add key="ClownFish_Aspnet_RequestBufferSize" value="2048"/>
    </appSettings>    
</configuration>

2048 这个数字就是缓冲区的大小，单位: byte

注意： 这个数字不能过大，否则会占用大量内存，甚至影响程序稳定！

建议值： 2048

请求缓冲区启用条件

请求缓冲区的开启需要几个条件：

• ClownFish_Aspnet_RequestBufferSize 的设置值 大于 0
• 存在请求体，例如： POST 是支持的，GET请求就不支持
• 请求体内容是文本格式，这里主要分析 Content-Type
• 请求体长度小于 ClownFish_Aspnet_RequestBufferSize 的设置值

以上4个条件必须全部满足才会启用。

确认请求缓冲区已开启

可以查看响应头，例如：

HTTP/1.1 200 OK
x-EnableBuffering: 2048

当存在 x-EnableBuffering 这个响应头时，就表示请求缓冲区已开启。

查看日志

例如：