一条占100字节
1 10 10_000 10_000_000
100 1k 1m 1G
一个节点10G的空间
一条redolog: orderItemId:redisId:queue:offset:skuID:skuNum:time 标记一次扣减
消息队列:
回退扣减可以做成一个延时任务,利用zset做消息队列;
商品扣减后, 一个消费者负责把redo_log同步到redis节点的zset里.这里实现限流措施.
10分钟后一个定时任务负责消费redo_log里面的redo_log
zset是有幂等性的,因此可以做.
一条占100字节
1 10 10_000 10_000_000
100 1k 1m 1G
一个节点10G的空间
一条redolog: orderItemId:redisId:queue:offset:skuID:skuNum:time:redisTerm:memberId 标记一次扣减
memberId用于路由
消息队列:
回退扣减可以做成一个延时任务,利用zset做消息队列;
商品扣减后, 一个消费者负责把redo_log同步到redis节点的zset里.这里实现限流措施.
10分钟后一个定时任务负责消费redo_log里面的redo_log
zset是有幂等性的,因此可以做.
1.mq将redolog同步到redis zset,达到容量限制就停止
2.定时任务根据time,决定是否需要关单,不需要就休息到指定时间.
同步至zset:
同步lua:
获取队列大小,超过100_000,就暂停同步,
将数据同步至zset
定时任务消费:
1.批量取出100条消息;对每一条消息:
a.查询订单支付状况,不是成功支付就返回库存
b.如果支付成功,查询扣减状况,扣减的orderItemId:redisID对应的redo_log与本次相同,那就不在返还库存
否则返还库存.
返还库存lua:
1.查询该队列的锁是否由自己占据:不是就返回 lock failed 锁信息是uuid
2.查看redisTerm,与当前相等才执行,否则返回 wrong term
3.查看sku对应的版本号,如果大于等于当前offset,就返回.表明已经执行过了 has been done
4.执行返还.
5.将该元素移除队列 success
根据返回结果:
lock failed ,终止当前批次的消费,重新获取自己被分配的
wrong term 表示消费成功, 跳过该消息
has been done 跳过该消息
success 消费成功,跳过该消息
failed 重试三次后,丢弃该消息;也就是消费成功
定时任务验证一致性:
每个redis节点运行一段时间后会进入休眠期,该休眠期是指定的,由任何一个节点运行时设置.比如运行时间为三个小时,执行完同步库存任务后,设置下一次到期时间.
同步Lua:
–查询当前redis节点是否在维护,如果在维护,就放弃本次同步
local maintain_time= redis.call(‘hget’,‘metadata’,‘next_maintain_time’)
time=tonumber(maintain_time)
if(time<curTime) return ‘MAINTAINING’
local zset_size= redis.call(‘ZCARD’,restock_queue_name)
local size=tonumber(zset_size) or -1
if size>100000 then
return ‘FULL’
end
if size!=-1 then
redis.call(‘zadd’,redo_log_queue_name,offset,restock_item)
return ‘SUCCESS’
end
消费Lua:
local maintain_time= redis.call(‘hget’,‘metadata’,‘next_maintain_time’)
time=tonumber(maintain_time)
if(time<curTime) return ‘MAINTAINING’
local lock=redis.call(‘hget’,‘restcok_lock’,restock_queue_name)
if lock == uuid then
local cur_term=redis.call(‘hget’,‘metadata’,‘cur_term’)
if cur_term == term then
local num:offset=redis.call(‘hget’,‘sku’,skuId)
cur_offset=tonumber(offset) or 0
if cur_offset > curOffset then
return ‘HAS_BEEN_CONSUMED’
else
local pre_num=tonumber(num) or 0
local newNum= pre_num + restockNum
local newValue=tostring(newNum)…‘:’…tostring(offset)
redis.call(‘hset’,‘sku’,skuId,newValue)
return ‘SUCCESS’
end
else
return ‘WRONG_TERM’
end
else
return ‘lock failed’
end
同步业务:
消息队列需要再次消费该业务;
消费业务:
定时任务,比较复杂,需要一个调度线程,先写核心业务
定时任务:
redis本身的restock队列消费.这个并不是定时任务,而是一个一直持续不断地任务,比较合适的情况:
调度线程负责派发任务,之后与执行线程通信,当调度线程认为某个线程下线时,需要重新派发该任务.
基于zookeeper是一个合适的方式.
redis重建任务:
这是个定时任务,但本质上也需要派发任务,监视该任务的执行情况,失败就要重新派发;
两种任务都可以多次执行,具有幂等性.
调度线程派发任务:
执行线程获取任务后执行任务,执行后释放该任务;
任务一zookeeper节点需要提供的信息:
redis ip:port restock_queue_name;
zookeeper节点需要提供的信息:
mysql ip:port redis ip:port 数据库的表名;
必须有监视集群,监视任务的执行状况,一旦断开就要重新派发任务.
哨兵节点派发任务:
工作节点向哨兵节点申请任务,哨兵节点派发任务(修改核心数据,并将任务详情发给工作节点),工作 节点需要向烧饼节点回复ACK,这样该任务就被占据了.之后哨兵节点开启心跳检测,保证工作节点处于存活状态.
断开连接时:
哨兵节点重新派发任务:
工作节点通知线程停止执行任务(给出一个共享变量表示状态),重新获取任务(这个简单),这里定时任务本身不需要分布式锁,本地锁就可以解决.
感觉用mysql,redis就行.
redis的阻塞队列负责分发任务,节点抢到以后就向哨兵节点发送消息,收到消息后版本号增1.心跳检测可以用数据库写来代替.哨兵节点定时扫描全表,获取很久没有更新的节点.嗯嗯这样就行;
第二个任务:
也是派发.派发成功后,修改Mysql,同步完成后.修改下一次到期时间.
数据库表设计:
redis_address:
mysql_address:
work_content:
type:
version:
work_owner:
next_maintain_time:
last_heart_beat_time:
work_status:
派发任务:
哪些任务需要投递:
1.没有worker的,心跳时间超时的,执行失败的,投递后没有认领的
1.全表扫描(只有抢到分布式锁才能执行).work_ownernull且sentinel_check_time到期的,扔进redis队列里.同时修改sentinel_check_time为current+1min
2.全表扫描 heart_beat_time+60s 小于当前时间的任务,扔进redis队列里.
worker从阻塞队列获取任务后,修改mysql, 将work_owner修改为自己的id,将heart_beat_time修改为当前时间,将work_status修改为正在执行,用版本号作为乐观锁.这里version,work_status两个作为乐观锁.同一个任务,先后被A,B获得
调度线程第一次派发任务, version=0 work_status=0,修改完数据库后将任务派发到redis里
A分到任务,写数据库 heart_beat_time=now(),work_status=1 where version=0 and work_status=0
A的写入受GC,网络影响出现延迟,调度线程再一次派发任务: version=1,work_status=0
B拿到任务,写数据库,
heart_beat_time=now(),work_status=1 where version=1 and work_status=0
如果B写入,A就不会成功写入
如果A成功,调度线程写入失败,B也不会拿到任务
调度线程随机选择一个元数据表:
1.从元数据表中选出一个任务表, 选择逻辑是 next_schedule_time<now
更新 next_schedule_time=now()+2min,term=term+1 where term=#{term} 提交事务
2.从任务表中选择未执行的业务: 派发超时,执行超时,没被派发的任务
a. heart_beat_time+30s<now and status1 没有被占据的业务
b. heart_beat_time+120s<now and status==2 心跳超时的业务
b. status=0 没有被执行的业务
将记录的version=term后,派发到阻塞队列里.对这里是乐观修改,使用任务表的version作为乐观锁,调度周期作为修改后的version值,cas(expect=version,update=term)
调度线程:
调度线程从元数据表获取要扫描的任务表;设置执行状态status=1(正在执行),执行时间=now(),下次执行时间=无穷大
扫描任务表,修改派发任务的数据后,将元数据表对应记录的下次执行时间设置为30s以后,status=2(执行完毕)
将任务派发到redis阻塞队列里
工作线程:
从阻塞队列获取任务,heart_beat_time=now,work_status=2(正在进行) version=#{term}
sql:heart_beat_time=#{time},work_status=2,version=#{term} where version=#{version} and work_status=0
失败就表示抢夺失败,那就再次阻塞等待.
之后开始执行任务, 并在另一个线程里定时汇报自己的心跳时间.
元数据表设计:
scheduled_task_metadata
id
term
next_schedule_time
task_table_name
任务表设计:
scheduled_task
version
heart_beat_time
task_status
task_type
task_content
work_type根据这个找实现类;
任务工作队列的逻辑:
每一个节点都有一个二级调度线程池,负责获取任务,派发给下面的工作线程
二级调度线程:
生产者锁
阻塞队列获取元素
扔到三级阻塞队列里.
三级工作线程:
从三级阻塞队列获取任务,将心跳检测任务派发给心跳检测线程池
while(flag)
执行任务,
心跳检测线程池: 每隔10s进行一次汇报,如果发现任务的version改变,将flag设置为false,通知工作线程结束工作
调度线程工作逻辑:
调度线程池本身是一个定时线程池,每隔20s扫描元数据表,这里抢一个有过期时间的分布式锁
select * from work_meta_data where next_schedule_time<= (#{curTime}-120_000) limit 1
update work_meta_data set term=#{newTerm}+1 and next_schedule_time=(#{curTime}+120_000) where id=#{id} and term=#{preTerm}
之后select 任务表,
select * from table_name where
a. heart_beat_time+30s<now and status==1 没有被占据的业务
b. heart_beat_time+120s<now and status==2 心跳超时的业务
b. status=0
用自己的term作为版本号,发送到redis阻塞队列里
