RedisTokenStore 源码解析 以及内存泄漏问题
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RedisTokenStore 源码解析 以及内存泄漏问题相关的知识,希望对你有一定的参考价值。
参考技术A 前端时间,正好在做公司权限相关的架构问题,然后选择了Spring OAuth2来作为公司权限框架,先记录下目前遇到原生问题吧,后续有时间再来整理这个框架的整体脉络;RedisTokenStore 主要是来做token持久化到redis的工具类
我们先来看下缓存到redis中有哪些key
ACCESS :用来存放 AccessToken 对象(登录的token值,还有登录过期时间,token刷新值)
AUTH_TO_ACCESS :缓存的也是AccessToken 对象,是可以根据用户名和client_id来查找当前用户的AccessToken
AUTH :用来存放用户信息(OAuth2Authentication),有权限信息,用户信息等
ACCESS_TO_REFRESH :可以根据该值,通过AccessToken找到refreshToken
REFRESH :用来存放refreshToken
REFRESH_TO_ACCESS :根据refreshToken来找到AccessToken
CLIENT_ID_TO_ACCESS :存放当前client_id有多少AccessToken
UNAME_TO_ACCESS :当没有做单点登录的话,可以使用该key,根据用户名查找当前用户有多少AccessToken可以使用
根据client_id和用户名,来获取当前用户的accessToken,如果缓存中的OAuth2Authentication已经过期,或者雷勇有变化,则会重新更新缓存;
缓存OAuth2AccessToken 和 OAuth2Authentication 对象,该方法主要在登录时调用,会把上面说的所有key值都缓存起来;
再看下,移除accessToken时
目前主要是看这几个方法,其他方法也挺简单的,主要是一些redis缓存操作的;
client_id_to_access 和 uname_to_access 使用的是set集合,众所周知redis的set集合的过期时间是按照整个key来设置的;
每次登陆时,会先根据client_id和用户名去缓存中查找是否有可使用的AccessToken,如果有则返回缓存中的值,没有则生成新的;
所以每次登陆都会往这两个集合中放入新的accessToken,如果当某个用户在AccessToken有效期内没有操作,则当前用户的登陆信息会被动下线,access 和 auth 中缓存的值都会过期,再次登陆时就查找不到了;
但是如果当前平台用户量不小,那么一直都会有人操作,client_id_to_access 这个集合就会一直续期,那么过期了的accessToken就会一直存在该集合中,且不会减少,造成内存泄漏;
1.自己实现TokenStore,修改client_id_to_access 数据结构
2.写个定时任务,定期扫描client_id_to_access ,清除掉已经过期的token
3.如果不需要知道当前有哪些用户登录,或者该功能已经用了其他方式实现的,可以直接去掉这两个redis key
canal 源码解析系列-store模块解析
引言
parser模块用来订阅binlog事件,然后通过sink投递到store。store模块用来执行最终的落库(基于内存),数据存储。
正文
核心接口是CanalEventStore
,目前只有一个实现类MemoryEventStoreWithBuffer
,这是一个基于内存buffer构建内存memory store。先来看下类图:
CanalStoreScavenge
接口是干啥的?CanalEventStore都继承了它。看下它的接口定义:
/**
* store空间回收机制,信息采集以及控制何时调用{@linkplain CanalEventStore}.cleanUtil()接口
*
* @author jianghang 2012-8-8 上午11:57:42
* @version 1.0.0
*/
public interface CanalStoreScavenge {
/**
* 清理position之前的数据
*/
void cleanUntil(Position position) throws CanalStoreException;
/**
* 删除所有的数据
*/
void cleanAll() throws CanalStoreException;
}
通过注释大概能看出来,这是专门抽象了一个接口来解决数据清理的问题。比如定时清理,满了之后清理,每次 ack 清理等。晚点再说清理的具体实现原理。
CanalEventStore
接口定义的方法很多,不过总体可以分为4类接口:
- put
- get
- ack
- rollback
接口定义如下:
public interface CanalEventStore<T> extends CanalLifeCycle, CanalStoreScavenge {
/**
* 添加一组数据对象,阻塞等待其操作完成 (比如一次性添加一个事务数据)
*/
void put(List<T> data) throws InterruptedException, CanalStoreException;
/**
* 添加一组数据对象,阻塞等待其操作完成或者时间超时 (比如一次性添加一个事务数据)
*/
boolean put(List<T> data, long timeout, TimeUnit unit) throws InterruptedException, CanalStoreException;
/**
* 添加一组数据对象 (比如一次性添加一个事务数据)
*/
boolean tryPut(List<T> data) throws CanalStoreException;
/**
* 添加一个数据对象,阻塞等待其操作完成
*/
void put(T data) throws InterruptedException, CanalStoreException;
/**
* 添加一个数据对象,阻塞等待其操作完成或者时间超时
*/
boolean put(T data, long timeout, TimeUnit unit) throws InterruptedException, CanalStoreException;
/**
* 添加一个数据对象
*/
boolean tryPut(T data) throws CanalStoreException;
/**
* 获取指定大小的数据,阻塞等待其操作完成
*/
Events<T> get(Position start, int batchSize) throws InterruptedException, CanalStoreException;
/**
* 获取指定大小的数据,阻塞等待其操作完成或者时间超时
*/
Events<T> get(Position start, int batchSize, long timeout, TimeUnit unit) throws InterruptedException,
CanalStoreException;
/**
* 根据指定位置,获取一个指定大小的数据
*/
Events<T> tryGet(Position start, int batchSize) throws CanalStoreException;
/**
* 获取最后一条数据的position
*/
Position getLatestPosition() throws CanalStoreException;
/**
* 获取第一条数据的position,如果没有数据返回为null
*/
Position getFirstPosition() throws CanalStoreException;
/**
* 删除{@linkplain Position}之前的数据
*/
void ack(Position position) throws CanalStoreException;
/**
* 删除指定seqId之前的数据
*
* @Since 1.1.4
*/
void ack(Position position, Long seqId) throws CanalStoreException;
/**
* 出错时执行回滚操作(未提交ack的所有状态信息重新归位,减少出错时数据全部重来的成本)
*/
void rollback() throws CanalStoreException;
}
在深入分析源码之前,要先理解EventStore的含义,EventStore是一个RingBuffer,有三个指针:Put、Get、Ack。
- Put: Canal Server从MySQL拉取到数据后,放到内存中,Put增加
- Get: 消费者(Canal Client)从内存中消费数据,Get增加
- Ack: 消费者消费完成,Ack增加。并且会删除Put中已经被Ack的数据
RingBuffer是Disruptor中的设计概念,这个有兴趣的可以查阅相关资料
put
方法的实现有很多种,但是核心都差不多,最终都是调用doPut
方法,我们来看其中一个put方法的源码:
public boolean put(List<Event> data, long timeout, TimeUnit unit) throws InterruptedException, CanalStoreException {
if (data == null || data.isEmpty()) {
return true;
}
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
//检查是否满足插入条件
if (checkFreeSlotAt(putSequence.get() + data.size())) {
doPut(data);
return true;
}
//判断是否已经超时,如果超时,则不执行插入操作,直接返回false
if (nanos <= 0) {
return false;
}
try {
//队列满了,等一会
nanos = notFull.awaitNanos(nanos);
} catch (InterruptedException ie) {
//如果一直等待到超时,都没有可用空间可以插入,notFull.awaitNanos会抛出InterruptedException
//超时之后,唤醒一个其他执行put操作且未被中断的线程(感觉就是自己不行了,让其他兄弟再试试)
notFull.signal(); // propagate to non-interrupted thread
throw ie;
}
}
} finally {
lock.unlock();
}
}
操作之前先加锁,MemoryEventStoreWithBuffer
使用的是ReentrantLock
,考虑到put操作的线程安全,这里加速是显而易见的。不过有个细节我们考虑下,为啥要把成员变量lock赋值给一个本地变量再进行后续的操作呢?
其实如果你看的源码比较多,会发现很多地方都有类似的操作。我个人理解,这是一个良好的编码习惯。它至少有两个好处:一是拷贝给方法的本地变量后,访问效率更高(一个是在堆上,一个是栈上)。另一个原因本地变量用了final修饰,可以保证在操作的过程中即使成员变量被修改了也不会影响到自己。
然后我们看到notFull
变量,还有一个变量叫notEmpty
,他们两个一般搭配使用:
// 阻塞put/get操作控制信号
private ReentrantLock lock = new ReentrantLock();
private Condition notFull = lock.newCondition();
private Condition notEmpty = lock.newCondition();
这是一个比较经典的使用场景,在这个场景中,通常有生产者和消费者两个角色,然后有一个共享缓冲区。生产者用于将消息放入缓冲区,消费者用于从缓冲区中取出消息。
当缓冲区已经满了,而此时生产者还想放入一个新的数据的时候,解决方法是让生产者此时进行休眠(notFull.await),等待消费者从缓冲区中取走了一个或者多个数据后再去唤醒它(notFull.signal)。
同样当缓冲区已经空了,而消费者还想去取数据,此时让消费者进行休眠(notEmpty.await()),等待生产者放入一个或者多个数据时再唤醒它(notEmpty.signal)。我们可以用一幅图来描述上述流程:
checkFreeSlotAt
方法是用来检查是否满足插入条件,看看它的实现:
/**
* 查询是否有空位
*/
private boolean checkFreeSlotAt(final long sequence) {
//1、检查是否足够的slot。入参的sequence值是:当前putSequence位置 + 新插入的event的记录数。
//结合文章中的图,其减去bufferSize不能大于ack位置,或者换一种说法,减去bufferSize不能大于ack位置。
final long wrapPoint = sequence - bufferSize;//队列的长度默认是16*1024
final long minPoint = getMinimumGetOrAck();//取get或者ack较小的
if (wrapPoint > minPoint) { // 刚好追上一轮
return false;
} else {
// 在bufferSize模式上,再增加memSize控制
//2、如果batchMode是MEMSIZE,继续检查是否超出了内存限制
if (batchMode.isMemSize()) {
final long memsize = putMemSize.get() - ackMemSize.get();
if (memsize < bufferSize * bufferMemUnit) {
return true;
} else {
return false;
}
} else {
return true;
}
}
}
注释写的比较详细了,不多说。
接下来看看核心方法doPut
:
/**
* 执行具体的put操作
*/
private void doPut(List<Event> data) {
long current = putSequence.get();//当前put的位置
long end = current + data.size();
// 先写数据,再更新对应的cursor,并发度高的情况,putSequence会被get请求可见,拿出了ringbuffer中的老的Entry值
for (long next = current + 1; next <= end; next++) {
//通过getIndex方法对next变量转换成正确的位置,设置到Event[]数组中,
// Event[]数组是环形队列的底层实现,其大小为bufferSize值,默认为16 * 1024。
// getindex通过与操作一个mask防止越界
entries[getIndex(next)] = data.get((int) (next - current - 1));
}
putSequence.set(end);//更新位置
// 记录一下gets memsize信息,方便快速检索
if (batchMode.isMemSize()) {
long size = 0;
for (Event event : data) {
size += calculateSize(event);
}
putMemSize.getAndAdd(size);
}
profiling(data, OP.PUT);//记录put时间,监控用
// tell other threads that store is not empty
notEmpty.signal();
}
总结下doPut方法,主要有4个操作:
- 的event数据赋值到Event[]数组上
- 更新ringbuffer上的put位置
- 如果是内存模式,计算新插入的event数据对内存的影响
- 调用notEmpty.signal()方法,通知buffer有数据了(get操作阻塞的线程可以被唤醒)
get
方法不说了,理解了put很容易理解get方法。我们来看下ack
方法:
public void ack(Position position) throws CanalStoreException {
cleanUntil(position, -1L);
}
前面我们提到了CanalStoreScavenge
接口,有两个方法:cleanUntil
和cleanAll
,后者在stop的时候调用,而前者就是在ack的时候调用。
public void cleanUntil(Position position, Long seqId) throws CanalStoreException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
long sequence = ackSequence.get();//当前ack的位置
long maxSequence = getSequence.get();//当前get的位置
boolean hasMatch = false;
long memsize = 0;
// ack没有list,但有已存在的foreach,还是节省一下list的开销
long localExecTime = 0L;
int deltaRows = 0;
if (seqId > 0) {
maxSequence = seqId;
}
//遍历所有未被ack的event,从中找出与需要ack的position相同位置的event,清空这个event之前的所有数据。
for (long next = sequence + 1; next <= maxSequence; next++) {
Event event = entries[getIndex(next)];//要ack的event
if (localExecTime == 0 && event.getExecuteTime() > 0) {
localExecTime = event.getExecuteTime();
}
deltaRows += event.getRowsCount();//要ack的event包含的mysql行数
memsize += calculateSize(event);//要ack的event占用字节数
if ((seqId < 0 || next == seqId) && CanalEventUtils.checkPosition(event, (LogPosition) position)) {
// 找到对应的position,更新ack seq
hasMatch = true;
if (batchMode.isMemSize()) {
ackMemSize.addAndGet(memsize);
// 尝试清空buffer中的内存,将ack之前的内存全部释放掉
for (long index = sequence + 1; index < next; index++) {
entries[getIndex(index)] = null;// 设置为null,方便GC
}
// 考虑getFirstPosition/getLastPosition会获取最后一次ack的position信息
// ack清理的时候只处理entry=null,释放内存
Event lastEvent = entries[getIndex(next)];
lastEvent.setEntry(null);//方便gc
lastEvent.setRawEntry(null);//方便gc
}
//更新ack的值
if (ackSequence.compareAndSet(sequence, next)) {// 避免并发ack
notFull.signal();//清空了就可以继续put数据了
ackTableRows.addAndGet(deltaRows);
if (localExecTime > 0) {
ackExecTime.lazySet(localExecTime);
}
return;
}
}
}
if (!hasMatch) {// 找不到对应需要ack的position
throw new CanalStoreException("no match ack position" + position.toString());
}
} finally {
lock.unlock();
}
}
Ack操作实际上就是将消费成功的事件从队列中删除,如果一直不Ack的话,队列满了之后,Put操作就无法添加新的数据了。
rollback
方法比较简单,这里不多说了。
参考:
- https://www.bookstack.cn/read/canal-v1.1.4/34357b71c7c1f182.md#%E6%9E%B6%E6%9E%84
- http://www.tianshouzhi.com/api/tutorials/canal/401
以上是关于RedisTokenStore 源码解析 以及内存泄漏问题的主要内容,如果未能解决你的问题,请参考以下文章
Java Executor源码解析—Executors线程池工厂以及四大内置线程池