Redis与Kafka达成高并发更新库存与数据一致性保证

Posted 2022-03-01 greatsharp

一. 并发更新方案

采用在Redis中判断并更新库存（库存值可增可减），由Redis保证库存的正确性，由Kafka与MongodDB数据库事务保证最终的一致性。

二. Redis使用lua脚本来保证并发更新的正确性

1. Redis会将整个lua脚本作为一个整体执行，中间不会被其它命令插入，保证了原子性，线程安全，且不需要事务控制。

2. Redis中的值，假设正确范围在0 <= x <= max（通过脚本传参）

3. 脚本的作用是，检测key指定的值与increment之和counter是否在正确范围内；如果在正确范围内则更新为counter，更新成功返回1，更新失败返回0；如果不在正确范围内则返回0。

4. Redis执行脚本成功，不论后续发送消息成功如否，都要给业务调用方返回成功。如果执行脚本失败，直接给业务调用方返回失败，无后续动作。

三. Kafka配置与消息主要内容

生产者采用异步发送加回调的方式，若发送失败或出现异常，则必须在回调函数里保证消息一定要发送成功。

生产者的acks设置为all，保证不丢消息。

消息内容：

key：long类型，kafka消息key，雪花算法唯一ID，

orderId：long类型，订单ID，雪花算法唯一ID，

skuId：long类型，库存商品ID，雪花算法唯一ID，

increment：int类型，增加或减少库存，

timestamp：long类型，时间戳

消费者配置手动提交ACK，消费消息的接口逻辑如下：

接口开启MongoDB事务（REQUIRED_NEW，rollbackFor=Exception.class）

1）插入防重表，MongoDB的集合id设置为消息的key或orderId都可以，保证唯一就行。

2）更新库存表，使用MongoDB的inc指令对库存值做原子操作。

3）返回更新结果。

因为MongoDB中的WriteConflict问题，上述接口需要重试直到保证MongoDB更新成功。消费接口调用成功后才能ACK。

重复消费的问题：

事务中需要插入防重表，重复的消息key会插入失败，MongoDB事务会回滚，保证了数据库中库存值的正确性。

四. 并发更新与补偿逻辑验证

假设库存值的正确范围在0至1000，即0 <= x <= 1000

时序	线程A+1	线程B-1	Redis值	DB值	备注
			999	999
	执行lua脚本+1		1000	999
	发送+1消息成功	执行lua脚本-1	999	1000
		发送-1消息成功	999	999	最终一致

时序	线程A+1	线程B-1	线程C+1	Redis值	DB值	备注
				999	999
	执行lua脚本+1成功			1000	999
	发送+1消息成功		执行lua脚本+1，超出范围失败	1000	1000
	给业务返回成功	执行lua脚本-1	给业务返回失败	999	1000
		发送-1消息成功		999	999
		给业务返回成功		999	999	最终一致

时序	线程A+1	线程B-1	线程C+1	线程D+1	Redis值	DB值	备注
					999	999
	执行lua脚本+1成功				1000	999
	发送+1消息成功		执行lua脚本+1，超出范围失败		1000	1000
	给业务返回成功	执行lua脚本-1 成功	给业务返回失败		999	1000
		发送-1消息失败		执行lua脚本+1 成功	1000	1000	消息发送失败，插入消息发送重试表，异步重试直到成功发送。
		给业务返回成功		发送+1消息成功	1000	1001
		异步发送-1消息最终成功		给业务返回成功	1000	1000	最终一致

时序	线程A+1	线程B-1	线程C+1	线程D+1	Redis值	DB值	备注
					999	999
	执行lua脚本+1 成功				1000	999
	发送+1消息成功		执行lua脚本+1， 超出范围失败		1000	1000
	给业务返回成功	执行lua脚本-1成功	给业务返回失败		999	1000
		发送-1消息失败		执行lua脚本+1 成功	1000	1000	补偿redis，incrby+1
		补偿incrby+1		发送消息+1	1001	1001
		给业务返回失败		给业务返回成功	1001	1001	超卖，说明发送消息失败不能补偿redis

说明：

1.如果用Redis保证正确性，Redis操作成功则必须给业务返回成功，后续出现错误或异常时是不能反向操作补偿Redis的。

2.异步发送消息失败时，回调方法里只能使用其它方式保证消息一定要发送成功。例如异步线程无限重试。