CAP.RabbitMQ.MongoDB示例

CAP
●CAP docs看到更多详细资料。
●CAP 视频教程，学习如何在项目中集成CAP。
●GitHub源码：https://github.com/dotnetcore/cap
●示例代码：https://github.com/dotnetcore/CAP/tree/master/samples

CAP 运输器

通过运输将数据从一个地方移动到另一个地方-在采集程序和管道之间，管道与实体数据库之间，甚至在管道与外部系统之间。

怎么选择运输器

RabbitMQ

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ 服务器是用 Erlang 语言编写的，而聚类和故障转移是构建在开源的通讯平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。
CAP 支持使用 RabbitMQ 作为消息传输器。

RabbitMQ Options

CAP 直接对外提供的 RabbitMQ 配置参数如下：

CustomHeaders Options

当需要从异构系统或者直接接收从RabbitMQ 控制台发送的消息时，由于 CAP 需要定义额外的头信息才能正常订阅，所以此时会出现异常。通过提供此参数来进行自定义头信息的设置来使订阅者正常工作。
你可以在这里找到有关 头信息 的说明。
用法如下：

config.UseRabbitMQ(o =>
{
    o.CustomHeaders = e => new List<KeyValuePair<string, string>>
    {
        new KeyValuePair<string, string>(Headers.MessageId, SnowflakeId.Default().NextId().ToString()),
        new KeyValuePair<string, string>(Headers.MessageName, e.RoutingKey),
    };
});

连接 RabbitMQ 集群

使用逗号分隔连接字符串即可，如下：

x=> x.UseRabbitMQ("localhost:5672,localhost:5673,localhost:5674")

MongoDB

MongoDB 是一个跨平台的面向文档型的数据库程序，它被归为 NOSQL 数据库，CAP 从 2.3 版本开始支持 MongoDB 作为消息存储。
MongoDB 从 4.0 版本开始支持 ACID 事务，所以 CAP 也只支持 4.0 以上的 MongoDB，并且 MongoDB 需要部署为集群，因为 MongoDB 的 ACID 事务需要集群才可以使用。
有关开发环境如何快速搭建 MongoDB 4.0+ 集群，你可以我的参考 这篇文章。

UseMongoDB配置项

RabbitMQ.MongoDB示例

NuGet

Install-Package DotNetCore.CAP.RabbitMQ
Install-Package DotNetCore.CAP.Dashboard
Install-Package DotNetCore.CAP.MongoDB
Install-Package Microsoft.Extensions.Configuration

注入CAP

using CAP.Transport.RabbitMQ.MongoDB;
using CAP.Transport.RabbitMQ.MongoDB.Models;
using DotNetCore.CAP.Internal;
using DotNetCore.CAP.Messages;
using MongoDB.Driver;
var builder = WebApplication.CreateBuilder(args);
// Add services to the container.
builder.Services.AddControllers();
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();
var appSetting = new AppSetting();
builder.Configuration.GetSection("AppSetting").Bind(appSetting);
//把AppSetting实体注入到容器,方便在构造函数里使用IOptionsSnapshot<AppSetting> options
builder.Services.Configure<AppSetting>(builder.Configuration.GetSection("AppSetting"));
//MongoDB 的客户端接口
builder.Services.AddSingleton<IMongoClient>(new MongoClient(appSetting.MongoDbSetting.Connection));
builder.Services.AddCap(config =>
{
    config.UseMongoDB(appSetting.MongoDbSetting.Connection);
    config.UseRabbitMQ(o =>
    {
        o.HostName = appSetting.RabbitMQSetting.HostName;
        o.UserName = appSetting.RabbitMQSetting.UserName;
        o.Password = appSetting.RabbitMQSetting.Password;
        o.Port = appSetting.RabbitMQSetting.Port;
        o.CustomHeaders = e => new List<KeyValuePair<string, string>>
        {
            new KeyValuePair<string, string>(Headers.MessageId, SnowflakeId.Default().NextId().ToString()),
            new KeyValuePair<string, string>(Headers.MessageName, e.RoutingKey),
        };
        o.ConnectionFactoryOptions = opt => {
            //rabbitmq client ConnectionFactory config
        };
    });//配置RabbitMQ
    config.UseDashboard();//配置Dashboard
    config.FailedRetryCount = 5;
    config.FailedThresholdCallback = failed =>
    {
        var logger = failed.ServiceProvider.GetService<ILogger<Program>>();
        logger?.LogError($@"A message of type {failed.MessageType} failed after executing {config.FailedRetryCount} several times,requiring manual troubleshooting. Message name: {failed.Message.GetName()}");
    };
});
var app = builder.Build();
// Configure the HTTP request pipeline.
if (app.Environment.IsDevelopment())
{
    app.UseSwagger();
    app.UseSwaggerUI();
}
app.UseHttpsRedirection();
app.UseAuthorization();
app.MapControllers();
app.Run();

公共部分

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.AspNetCore": "Warning"
    }
  },
  "AllowedHosts": "*",
  "AppSetting": {
    "SqlServerSetting": {
      "Connection": "Server=192.168.3.40;Database=webdemo;User=sa;Password=longfuchu;"
    },
    "MysqlSetting": {
      "Connection": "server=192.168.3.40;port=3306;user=root;password=123456;database=webdemo;charset=utf8;Allow Zero Datetime=true;sslmode=none;Old Guids=true;Allow User Variables=True"
    },
    "RedisSetting": {
      "Connection": "192.168.3.40:6379,password=",
      "Database": "test"
    },
    "RabbitMQSetting": {
      "HostName": "localhost",
      "UserName": "admin",
      "Password": "admin",
      "Port": 5672
    },
    "MongoDbSetting": {
      "Connection": "mongodb://192.168.3.40:2717", ///?replicaSet=rs0
      "Database": "test"
    }
  }
}

namespace CAP.Transport.RabbitMQ.MongoDB.Models;
/// <summary>
/// mongodb setting
/// </summary>
public class AppSetting
{
    /// <summary>
    /// SqlServerSetting
    /// </summary>
    public SqlServerSetting SqlServerSetting { get; set; }
    /// <summary>
    /// MysqlSetting
    /// </summary>
    public MysqlSetting MysqlSetting { get; set; }
    /// <summary>
    /// RedisSetting
    /// </summary>
    public RedisSetting RedisSetting { get; set; }
    /// <summary>
    /// RabbitMQSetting
    /// </summary>
    public RabbitMQSetting RabbitMQSetting { get; set; }
    /// <summary>
    /// MongoDbSetting
    /// </summary>
    public MongoDbSetting MongoDbSetting { get; set; }
    /// <summary>
    /// KafkaSetting
    /// </summary>
    public KafkaSetting KafkaSetting { get; set; }
}
public class RabbitMQSetting
{
    /// <summary>
    /// HostName
    /// </summary>
    public string HostName { get; set; } = "localhost";
    /// <summary>
    /// Password
    /// </summary>
    public string Password { get; set; }
    /// <summary>
    /// Username
    /// </summary>
    public string UserName { get; set; }
    /// <summary>
    /// The port to connect on.
    /// </summary>
    public int Port { get; set; } = 5672;
}
public class RedisSetting
{
    /// <summary>
    /// connection string
    /// </summary>
    public string Connection { get; set; }
    /// <summary>
    /// database name
    /// </summary>
    public string Database { get; set; }
}
public class MongoDbSetting
{
    /// <summary>
    /// connection string
    /// </summary>
    public string Connection { get; set; }
    /// <summary>
    /// database name
    /// </summary>
    public string Database { get; set; }
}
public class SqlServerSetting
{
    /// <summary>
    /// connection string
    /// </summary>
    public string Connection { get; set; }
}
public class MysqlSetting
{
    /// <summary>
    /// connection string
    /// </summary>
    public string Connection { get; set; }
}
public class KafkaSetting
{
    /// <summary>
    /// Servers
    /// </summary>
    public string Servers { get; set; }
}

using Microsoft.AspNetCore.Mvc;
namespace CAP.Transport.RabbitMQ.MongoDB.Controllers;
/// <summary>
/// 基础控制器
/// </summary>
[Route("api/[area]/[controller]/[action]")]
[ApiController]
public abstract class BaseController : ControllerBase
{
}

using Microsoft.AspNetCore.Mvc;
namespace CAP.Transport.RabbitMQ.MongoDB.Controllers.MongoDB;
/// <summary>
/// 域控制器
/// </summary>
[Area("MongoDB")]
public abstract class AreaController : BaseController
{
}

发布消息

using CAP.Transport.RabbitMQ.MongoDB.Models;
using DotNetCore.CAP;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Options;
using MongoDB.Bson;
using MongoDB.Driver;
using MongoDB.Driver.Core.Configuration;
namespace CAP.Transport.RabbitMQ.MongoDB.Controllers.MongoDB;
/// <summary>
/// 发送消息
/// </summary>
public class PublishController : AreaController
{
    private readonly IMongoClient _client;
    private readonly ICapPublisher _capBus;
    private readonly MongoDbSetting _mongoDbSetting;
    public PublishController(IMongoClient client, ICapPublisher capBus, IOptionsSnapshot<AppSetting> options)
    {
        _client = client;
        _capBus = capBus;
        _mongoDbSetting = options != null ? options.Value.MongoDbSetting:new MongoDbSetting();
    }
    /// <summary>
    /// 发布者-无事务
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public IActionResult WithoutTransaction()
    {
        _capBus.PublishAsync("sample.rabbitmq.mongodb", DateTime.Now);
        return Ok();
    }
    /// <summary>
    /// 发布者-发布不自动提交
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public IActionResult PublishNotAutoCommit()
    {
        //注意：MongoDB 不能在事务中创建数据库和集合，所以你需要单独创建它们，模拟一条记录插入则会自动创建
        var mycollection = _client.GetDatabase("test").GetCollection<BsonDocument>("test.collection");
        mycollection.InsertOne(new BsonDocument { { "test", "test" } });
        using (var session = _client.StartTransaction(_capBus, autoCommit: false))
        {
            //var collection = _client.GetDatabase("test").GetCollection<BsonDocument>("test.collection");
            //collection.InsertOne(session, new BsonDocument { { "hello", "world" } });
            _capBus.Publish("sample.rabbitmq.mongodb", DateTime.Now);
            session.CommitTransaction();
        }
        return Ok();
    }
    /// <summary>
    /// 发布者--
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public IActionResult PublishWithoutTrans()
    {
        //注意：MongoDB 不能在事务中创建数据库和集合，所以你需要单独创建它们，模拟一条记录插入则会自动创建
        var mycollection = _client.GetDatabase("test").GetCollection<BsonDocument>("test.collection");
        mycollection.InsertOne(new BsonDocument { { "test", "test" } });
        using (var session = _client.StartTransaction(_capBus, autoCommit: true))
        {
            //var collection = _client.GetDatabase("test").GetCollection<BsonDocument>("test.collection");
            //collection.InsertOne(session, new BsonDocument { { "hello", "world" } });
            _capBus.Publish("sample.rabbitmq.mongodb", DateTime.Now);
        }
        return Ok();
    }
}

处理消息

using DotNetCore.CAP;
using DotNetCore.CAP.Messages;
using Microsoft.AspNetCore.Mvc;
namespace CAP.Transport.RabbitMQ.MongoDB.Controllers.MongoDB;
/// <summary>
/// 处理消息
/// </summary>
[ApiController]
public class ConsumerController : ControllerBase
{
    /// <summary>
    /// 订阅消费者-
    /// </summary>
    /// <param name="value"></param>
    [NonAction]
    [CapSubscribe("sample.rabbitmq.mongodb")]
    public void ReceiveMessage(DateTime time)
    {
        Console.WriteLine($@"{DateTime.Now}, Subscriber invoked, Sent time:{time}");
    }
}

测试

测试报错没信息:”Standalone servers do not support transactions.”