分布式事务之消息补偿解决方案

一、数据库本地事务

先看看数据库事务的定义：单个逻辑工作单元执行的一系列操作，要么完全地执行，要么完全地不执行

这个比较容易理解，操作过数据库的一般都懂，既是业务需求涉及到多个数据表操作的时候，需要用到事务

要么一起更新，要么一起不更新，不会出现只更新了部分数据表的情况，下边看看数据库事务的使用

begin tran
begin try
update Table1 set Field = 1 where ID = 1
update Table2 set Field = 2 where ID = 1
end try
begin catch
rollback tran
end catch
commit tran

上实例在小型项目中一般是问题不大的，因为小型项目一般是单机系统，数据库、Web服务大都在一台服务器上，甚至可能只有一个数据库文件，

这种情况下使用本地事务没有一点问题；

但是本地事务有很大的缺陷，因为开启事务一般是锁表的，事务执行期间会一直锁着，其他的操作一般都要排队等待，对性能要求比较高的系统是不能忍受的。

特别是涉及改动不同数据库的操作，这会造成跨库事务，性能更加低

如果还涉及到不在同一台服务器、甚至不同网段部署的数据库，那本地事务简直是系统运行的灾难，是首先需要丢弃的解决方案。



那如果遇到上述情况，该怎么做呢，这就涉及到分布式事务了

二、分段式事务的补偿机制

如果有海量数据需要处理、或者要求高并发请求的话，同步的事务机制已经是不现实的了，这种情况下必须采用异步事务机制，既分段式的事务

分段式事务一般做法就是把需求任务分段式地完成，通过事务补偿机制来保证业务最终执行成功，补偿机制一般可以归类为2种：

1 ）定时任务补偿：

　　通过定时任务去跟进后续任务，根据不同的状态表确定下一步的操作，从而保证业务最终执行成功，

　　这种办法可能会涉及到很多的后台服务，维护起来也会比较麻烦，这是应该是早期比较流行的做法

2） 消息补偿：

　　通过消息中间件触发下一段任务，既通过实时消息通知下一段任务开始执行，执行完毕后的消息回发通知来保证业务最终完成；

　　当然这也是异步进行的，但是能保证数据最终的完整性、一致性，也是近几年比较热门的做法

定时任务补偿就不说了，这篇文章我们来讨论一下通过消息补偿来完成分布式事务的一般做法

三、分布式事务之消息补偿

0）我们以简单的产品下单场景来说明，（不要较真哈）

1）先来看看分布式异步事务处理流程示意图，APP1与APP2需要互相订阅对方消息



2）首先看数据库，2个，一个库存库，一个已下单成功的库

-- 下单通知，主要作用保留已下单操作，消息发送失败可以根据此表重新发送
CREATE TABLE [dbo].[ProductMessage](
[ID] [int] IDENTITY(1,1) NOT NULL,
[Product] [varchar](50) NULL,
[Amount] [int] NULL,
[UpdateTime] [datetime] NULL
)
-- 库存
CREATE TABLE [dbo].[ProductStock](
[ID] [int] IDENTITY(1,1) NOT NULL,
[Product] [varchar](50) NULL,
[Amount] [int] NULL
)
-- 下单成功
CREATE TABLE [dbo].[ProductSell](
[ID] [int] IDENTITY(1,1) NOT NULL,
[Product] [varchar](50) NULL,
[Customer] [int] NULL,
[Amount] [int] NULL
)
-- 下单成功消息，主要作用防止重复消费
CREATE TABLE [dbo].[ProductMessageApply](
[ID] [int] IDENTITY(1,1) NOT NULL,
[MesageID] [int] NULL,
[CreateTime] [datetime] NULL
)

3）项目架构Demo



数据底层访问使用的是Dapper、使用redis作为消息中间件

4）实体层代码

public class ProductMessage
{
[Key]
[IgnoreProperty(true)]
public int ID { get; set; }
public string Product { get; set; }
public int Amount { get; set; }
public DateTime UpdateTime { get; set; }
}
public class ProductMessageApply
{
[Key]
[IgnoreProperty(true)]
public int ID { get; set; }
public int MesageID { get; set; }
public DateTime CreateTime { get; set; }
}
public class ProductSell
{
[Key]
[IgnoreProperty(true)]
public int ID { get; set; }
public string Product { get; set; }
public int Customer { get; set; }
public int Amount { get; set; }
}
public class ProductStock
{
[Key]
[IgnoreProperty(true)]
public int ID { get; set; }
public string Product { get; set; }
public int Amount { get; set; }
}

5）服务接口层代码

public interface IProductMessageApplyService
{
void Add(ProductMessageApply entity);
ProductMessageApply Get(int id);
}
public interface IProductMessageService
{
void Add(ProductMessage entity);
IEnumerable<ProductMessage> Gets(object paramPairs = null);
void Delete(int id);
}
public interface IProductSellService
{
void Add(ProductSell entity);
}
public interface IProductStockService
{
void ReduceReserve(int id, int amount);
}

6）库存、消息通知

public class ProductMessageService : IProductMessageService
{
private IRepository<ProductMessage> repository;

public ProductMessageService(IRepository<ProductMessage> repository)
{
this.repository = repository;
}

public void Add(ProductMessage entity)
{
this.repository.Add(entity);
}

public IEnumerable<ProductMessage> Gets(object paramPairs = null)
{
return this.repository.Gets(paramPairs);
}

public void Delete(int id)
{
this.repository.Delete(id);
}
}

public class ProductStockService : IProductStockService
{
private IRepository<ProductStock> repository;

public ProductStockService(IRepository<ProductStock> repository)
{
this.repository = repository;
}

public void ReduceReserve(int id, int amount)
{
var entity = this.repository.Get(id);
if (entity == null) return;

entity.Amount = entity.Amount - amount;
this.repository.Update(entity);
}
}

7）下单、下单成功消息

public class ProductMessageApplyService : IProductMessageApplyService
{
private IRepository<ProductMessageApply> repository;

public ProductMessageApplyService(IRepository<ProductMessageApply> repository)
{
this.repository = repository;
}

public void Add(ProductMessageApply entity)
{
this.repository.Add(entity);
}

public ProductMessageApply Get(int id)
{
return this.repository.Get(id);
}
}

public class ProductSellService : IProductSellService
{
private IRepository<ProductSell> repository;

public ProductSellService(IRepository<ProductSell> repository)
{
this.repository = repository;
}

public void Add(ProductSell entity)
{
this.repository.Add(entity);
}
}

8）下单减库存测试

namespace Demo.Reserve.App
{
class Program
{
static void Main(string[] args)
{
Console.WriteLine(string.Format("{0} 程序已启动", DateTime.Now.ToString()));

Send();
Subscribe();

Console.ReadKey();
}

private static void Send()
{
var unitOfWork = new UnitOfWork(Enums.Reserve);

try
{
var productStockRepository = new BaseRepository<ProductStock>(unitOfWork);
var productStockServic = new ProductStockService(productStockRepository);
var productMessageRepository = new BaseRepository<ProductMessage>(unitOfWork);
var productMessageService = new ProductMessageService(productMessageRepository);

var id = 1;
var amount = 2;
var productMessage = new ProductMessage()
{
Product = "ProductCode",
Amount = amount,
UpdateTime = DateTime.Now
};

productStockServic.ReduceReserve(id, amount);
productMessageService.Add(productMessage);
unitOfWork.Commit();
Console.WriteLine(string.Format("{0} 减库存完成", DateTime.Now.ToString()));
Thread.Sleep(1000);

var message = JsonConvert.SerializeObject(productMessage);
RedisConfig.Instrace.Publish("channel.Send", message);
Console.WriteLine(string.Format("{0} 发送减库存消息: {1}", DateTime.Now.ToString(), message));
}
catch (Exception ex)
{
//Logger.Error(ex);
unitOfWork.Rollback();
}
}

private static void Subscribe()
{
var client = RedisConfig.Instrace.NewClient();
var subscriber = client.GetSubscriber();

subscriber.Subscribe("channel.Success", (chl, message) =>
{
try
{
var unitOfWork = new UnitOfWork(Enums.Reserve);
var productMessageRepository = new BaseRepository<ProductMessage>(unitOfWork);
var productMessageService = new ProductMessageService(productMessageRepository);

var messageID = message.ToString().ToInt();
if (messageID > 0)
{
productMessageService.Delete(messageID);
Console.WriteLine(string.Format("{0} 收到消费成功消息:{1}", DateTime.Now.ToString(), message));
}
}
catch (Exception ex)
{
//Logger.Error(ex);
}
});
}
}
}

9）下单成功及消息回发测试

namespace Demo.Sell.App
{
class Program
{
static void Main(string[] args)
{
Subscribe();

Console.WriteLine(string.Format("{0} 程序已启动", DateTime.Now.ToString()));
Console.ReadKey();
}

private static void Subscribe()
{
var client = RedisConfig.Instrace.NewClient();
var subscriber = client.GetSubscriber();

subscriber.Subscribe("channel.Send", (chl, message) =>
{
Consume(message);
});
}

private static void Consume(string message)
{
var unitOfWork = new UnitOfWork(Enums.Sell);

try
{
Console.WriteLine(string.Format("{0} 收到减库存消息: {1}", DateTime.Now.ToString(), message));

var productMessage = JsonConvert.DeserializeObject<ProductMessage>(message);

var productSellRepository = new BaseRepository<ProductSell>(unitOfWork);
var productSellService = new ProductSellService(productSellRepository);

var productMessageApplyRepository = new BaseRepository<ProductMessageApply>(unitOfWork);
var productMessageApplyService = new ProductMessageApplyService(productMessageApplyRepository);

var noExists = productMessageApplyService.Get(productMessage.ID) == null;
if (noExists)
{
productSellService.Add(new ProductSell()
{
Product = productMessage.Product,
Amount = productMessage.Amount,
Customer = 123
});

productMessageApplyService.Add(new ProductMessageApply()
{
MesageID = productMessage.ID,
CreateTime = DateTime.Now
});

unitOfWork.Commit();
Console.WriteLine(string.Format("{0} 消息消费完成", DateTime.Now.ToString()));
Thread.Sleep(1000);
}

RedisConfig.Instrace.Publish("channel.Success", productMessage.ID.ToString());
Console.WriteLine(string.Format("{0} 发送消费完成通知:{1}", DateTime.Now.ToString(), productMessage.ID.ToString()));
}
catch (Exception ex)
{
//Logger.Error(ex);
unitOfWork.Rollback();
}
}
}
}

10）好了，到了最后检验成果的时候了

先打开Demo.Sell.App.exe、然后打开Demo.Reserve.App.exe



大功告成！