CAP与动态数据库

[Gu17zz]

嗨，杨老师！
看了很多问答以及找了很多大家提出来的问题，比如#699 以及 #970，但是好像没有找到我想要的解决方案，只能在此提出我的问题了。
我这个项目中，cap消费是backgroundservice，生产是webapi，分了主库与多个分库（场景是主控数据库与多个子库），DbContext 实例分了两个：MasterDbContext、TranDbContext，其中 TranDbContext 是动态的。当使用 TranDbContext 下xxxxRepository.UnitOfWorkForCompanyDb.BeginTransaction(capPublisher, false) 时，这个 xxxxRepository 对应的可能都是不同的子库。最开始的报错是未找到[cap].[Published] 表，因为我只在AddCap中启用了MasterDbContext的UseEntityFramework()。于是后面我是用了个循环去开启子库的cap，如下：

                                services.AddCap(x =>
				{
					var connStrTemplate = configuration.GetEncryptedConnectionString(DbConnectionStringConstant.ConnectionStringTemplate);
					if (string.IsNullOrWhiteSpace(connStrTemplate))
					{
						logger.LogError($"{nameof(AddServicesExtensions)} --> Please set the [ConnectionStringTemplate] in appsettings.json");
						throw new ApplicationException($"Please set the [ConnectionStringTemplate] in appsettings.json");
					}

				        foreach (var item in locations)
					{
						var connStr = connStrTemplate
							.Replace("@server", item.database_server_ip)
							.Replace("@dbname", item.database_name)
							.Replace("@uid", item.database_login_id)
							.Replace("@pwd", item.database_login_pwd);

						// tran db---分库
						x.UseSqlServer(connStr);
					}

					// master db---主库
					x.UseSqlServer(configuration.GetEncryptedConnectionString(DbConnectionStringConstant.DefaultConnection));

					//CAP支持 RabbitMQ、Kafka、AzureServiceBus 等作为MQ，根据使用选择配置：
					var mqConfigure = configuration.GetSection("RabbitMQ").Get<RabbitMQOptions>();
					x.UseRabbitMQ(o =>
					{
						o.HostName = mqConfigure.HostName;
						o.Port = mqConfigure.Port;
						o.ExchangeName = mqConfigure.ExchangeName;
						o.VirtualHost = mqConfigure.VirtualHost;
						o.UserName = mqConfigure.UserName;
						o.Password = mqConfigure.Password;
					});

					//设置处理成功的数据在数据库中保存的时间（秒），为保证系统性能，数据会定期清理。
					x.SucceedMessageExpiredAfter = 24 * 3600;
					//设置失败重试次数
					x.FailedRetryCount = 50;
					//消息处理失败后的回调函数。FailedCallback 的类型为 Action<MessageType,string,string>，第一个参数为消息类型（发送的还是接收到），第二个参数为消息的名称（name），第三个参数为消息的内容（content）
					x.FailedThresholdCallback = (x)=>{
						logger.LogError($"{nameof(AddServicesExtensions)} --> FailedThresholdCallback --> ServiceProvider:{x.ServiceProvider},MessageType:{x.MessageType},Message:{x.Message}"); 
					};
					//x.DefaultGroupName = $"cap.queue.{Assembly.GetEntryAssembly()?.GetName().Name.ToLower() ?? "fin.mqbackgroundservice"}";
				});

但是这样并不行，后来我也尝试了 您在 #998 中发布的 SqlServerRabbitMqCapPublisher.cs 方案。这种动态多库的场景中，请问您有什么好的解决方案吗？

我现在是把cap消息记录表都放在了主控数据库（MasterDbContext）中，然后有些情况下，单独为了cap去开启了两个事务：

	                       using (var tranMaster = locationRepository.UnitOfWork.BeginTransaction(capPublisher, false))//--主库
				{
					using (var tran = await invoiceRepository.UnitOfWorkForCompanyDb.BeginTransactionAsync(tranMaster))//分库
					{
						logger.LogInformation($"Start the transaction to create invoice,inv ref:[{invoiceMasterHis.inv_ref}]");
                                                  
                                                  .............................
						
						await locationRepository.UnitOfWork.CommitTransactionAsync();

						await invoiceRepository.UnitOfWorkForCompanyDb.CommitTransactionAsync();
					}
				}

这里的tranMaster 仅仅只是为了cap而开启，如果改成使用子库去开启BeginTransaction(capPublisher, false)

invoiceRepository.UnitOfWorkForCompanyDb.BeginTransactionAsync(capPublisher, false)

则会报 [cap].[Published] 不存在了，所以挺郁闷的。我也跟我的队友们讨论过，我也向他们提出了我的质疑是： CAP不就是为了解决分布式事务的吗，为什么还要开启两个跨库事务。。

[yang-xiaodong]

建议查看这篇文章：https://www.cnblogs.com/savorboard/p/distributed-system-transaction-consistency.html

[Gu17zz]

建议查看这篇文章：https://www.cnblogs.com/savorboard/p/distributed-system-transaction-consistency.html

非常感谢！我会再去认真看一下的。