分布式、锁、延时任务

news2025/2/26 23:02:27

1. redission

2.zk

2.1 指令

ls / / 下有哪些子节点
get /zookeeper 查看某个子节点内容
create /aa “test”
delete /aa
set /aa “test01”

2.2 创建节点

模式默认创建永久
create -e 创建临时
create -e /zz “hello zz”
create -s 创建有序节点

create -s -e 临时序列化节点

2.3 节点事件监听

一次性的监听

节点创建事件监听 NodeCreated
stat -w /xx
节点删除监听 NodeDeleted
stat -w
节点数据变化监听 NodeDataChanged
get -w /bb
set /bb “cc” //修改节点数据
子节点监听 NodeChildrenChanged
ls -w /bb

create /bb/cc “test”

2.4 zk 分布式锁

独占、排它，其他线程获取不到
阻塞解决自旋的消耗 临时有序节点 + 事件监听(监听比他小1的节点)，

让最小的节点获取到锁, 44节点监听43节点，43 节点删除后，唤醒后续44节点获得锁。公平锁
3. 锁的可重入性
思路1：在节点的内容中记录服务器、线程、已经重入信息
思路2 ：Threadloacal：线程的局部变量，线程私有

阻塞操作获取不到监听并阻塞

创建 临时有序节点返回节点路径
// 获取前置节点，如果前置节点为空，那么获得锁成功，否则监听 前置节点

countdownLatch(1)
在监听事件的内部进行一个countDown();

可重入锁

threadloacal
在 trylock()

unlock 里面判断：

2.5 锁

zk 的节点 存储机制+通知机制
zk 有4中节点类型，持久节点，持久顺序节点、临时节点、临时顺序节点
持久和临时判断标准依赖于客户端的生命周期

顺序：节点对应的id

持久节点
- 持久顺序节点
- 持久的非顺序节点
临时节点
- 临时的非顺序节点( 实现分布式锁的条件)
- 临时的顺序节点

利用zk 支持的临时顺序节点 +通知机制 可以实现分布式锁

加锁：判断是否持有了分布式锁，判断客户端创建的节点是否是 有序节点中序号最小的一个
释放锁: 将自己的创建的节点删除即可。

3. redission zk 分布式锁对比

4. 延时任务

4.0 场景

4.1 定时任务数据库轮询

定时任务轮询数据库，
采用定时任务失效延迟、对业务表进行轮询判断，到点执行。有一点点误差。
1）使用单机版本的spring schedule +分布式锁的实现。
同一个服务，多个节点同时执行 可能都会执行，需要加分布式锁。

分布式调度框架 xxl-job
如果处理数据量较大，可以利用分布式调度系统的分片功能并行处理，大大提升数据的处理能力，加快处理速度。

优点：
单机版基于spring，实现简单。不用引入各种中间件，各个模块可以自行定义延迟规则。
缺点
2. 完全由业务代码进行控制，重复代码多，不论是否有待执行的数据，都要空轮询cpu 且频繁的访问数据库 io 消耗。
3. 由于是定时轮询，存在一点误差。

基于@schedule

取消订单操作

基于xxl-job 的

4.2 Java 的DelayQueue

Java DelayQueue的使用及应用场景

DelayQueue 的实现原理。
1） DelayQueue 是JDK提供的一个无界 BlockQueue ，用于放置实现了Delayed 接口的对象。其中的对象只能在其到期时，才能从队列中拿走。
这种队列是有序的，即队头对象的延迟到期时间最长。
注意不能将null 元素放置到这种队列中。

2）实现注意事项
队列里面的元素需要实现Delayed 这个接口。getDelay 方法用于设置延迟时间。 compareTo 方法用于对队列的元素 进行排序。

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    implements BlockingQueue<E> 
// 入队
put(){
offer();} 线程安全  reentrantlock 加锁 
offer(); 线程安全
// 出队 
poll(); 非阻塞方法，没有到期元素 直接返回 null
take(); 阻塞式 获取，没有到期元素，线程将会进行等待。

优点:
jdk 自带的，不需要引入其他框架中间件，实现简单。
缺点:

不支持分布式或者持久化的，重启会丢失。
如果并发量非常大，因为DelayQueue式无界的，队列内的对象越多。可能会造成oom 的风险。
所以使用 delayQueue实现的定时任务，只适用于任务量较小的情况。

4.3 消息中间件

rocketmq如何实现延时队列

rocketmq 先把消息按照延时时间段(1s,5s,10s) 发到指定的队列中,然后通过一个定时器轮询这些队列。如果到期，就把这个消息发到指定的topic 队列。
注意点:

rocketmq 延时消息的时长不支持随机时长的延迟。是通过特定的延迟等级来指定的。
默认支持18个等级的延迟消息。
延时等级在rocketmq服务端的MessageStoreConfig类中

private String messageDelayLevel ='1s 5s 10s 30s   
		1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 30m 1h 2h'
		发消息时： 设置delayLevel 等级即可  msg.setDelayLevel( level);
		level ==0 消息非延迟
		1<=level <=maxlevel 消息延迟特定的时间。
		level >maxlevel  那么 level=maxlevel  例如leve=20，延迟2h

优点：
基于消息中间件可以快速实现延时队列，而且天然支持消息消费的有序性、消息持久化、ack机制。
缺点：
需要额外的部署和运维成本。

4.4 redis zset、 key 过期回调

zrangebyScore

redis 127.0.0.1:6379> ZADD salary 2500 jack               
         # 测试数据
(integer) 0
redis 127.0.0.1:6379> ZADD salary 5000 tom
(integer) 0
redis 127.0.0.1:6379> ZADD salary 12000 peter
(integer) 0

redis 127.0.0.1:6379> ZRANGEBYSCORE salary -inf +inf           
    # 显示整个有序集
1) "jack"
2) "tom"
3) "peter"

redis 127.0.0.1:6379> ZRANGEBYSCORE salary -inf +inf WITHSCORES  
  # 显示整个有序集及成员的 score 值
1) "jack"
2) "2500"
3) "tom"
4) "5000"
5) "peter"
6) "12000"

redis 127.0.0.1:6379> ZRANGEBYSCORE salary -inf 5000 WITHSCORES  
  # 显示工资 <=5000 的所有成员
1) "jack"
2) "2500"
3) "tom"
4) "5000"

redis 127.0.0.1:6379> ZRANGEBYSCORE salary (5000 400000     
       # 显示工资大于 5000 小于等于 400000 的成员
1) "peter"