非阻塞同步算法实战（四）- 计数器定时持久化
问题背景及要求
- 需要对评论进行点赞次数和被评论次数进行统计，或者更多维度
- 要求高并发、高性能计数，允许极端情况丢失一些统计次数，例如宕机
- 评论很多，不能为每一个评论都一直保留其计数器，计数器需要有回收机制
问题抽象及分析
根据以上需求，为了方便编码与测试，我们把需求转化为以下接口
/**
* 计数器
*/
public interface Counter {
/**
* 取出统计数据，用Saver去持久化(仅定时器会调用，无并发)
* @param saver
*/
void save(Saver saver);
/**
* 计数(有并发)
* @param key 业务ID
* @param like 点赞
* @param comment 评论
*/
void add(String key, int like, int comment);
/**
* 持久化器，将数量持久化到数据库等
*/
@FunctionalInterface
interface Saver{
void save(String key, int like, int comment);
}
}简单分析可知，计数器比较简单，用AtomicInteger便能保证原子性，但考虑到计数器会被回收，则可能会出现这样的场景：某计数器已被回收了，此时继续在该计数器上计数，便会造成数据丢失，因此要处理该并发问题
解决方案
方案一
使用原生锁来解决竞争问题
/**
* 直接对所有操作上锁，来保证线程安全
*/
public class SynchronizedCounter implements Counter{
private HashMap<String, Adder> map = new HashMap<>();
@Override
public synchronized void save(Saver saver) {
map.forEach((key, value)->{//因为已加锁，所以可以安全地取数据
saver.save(key, value.like, value.comment);
});
map = new HashMap<>();
}
@Override
public synchronized void add(String key, int like, int comment) {
//因为已加锁，所以可以安全地更新数据
Adder adder = map.computeIfAbsent(key, x -> new Adder());
adder.like += like;
adder.comment += comment;
}
static class Adder{
private int like;
private int comment;
}
}方案点评：该方案让业务线程和定时保存线程竞争同一把实例锁，让他们互斥地访问，解决了竞争问题，但锁粒度太粗爆，性能低下
方案二
为了循序渐进，我们把“计数器需要有回收机制”这条要求去掉，这样我们可以很容易地利用上AtomicInteger这个类
/**
* 不回收计数器，问题变得简单许多
*/
public class IncompleteCounter implements Counter {
private ConcurrentHashMap<String, Adder> map = new ConcurrentHashMap<>();
@Override
public void save(Saver saver) {
map.forEach((key, value)->{//利用了AtomicInteger的原子特性，可以线程安全地取出所有计数，并置0(因为还会继续使用)
saver.save(key, value.like.getAndSet(0), value.comment.getAndSet(0));
});
//因为不回收，所以不用考虑Adder被回收丢弃后，仍被其它线程使用的情况(因为没有锁，所以这种情况是可能发生的)
}
@Override
public void add(String key, int like, int comment) {
Adder adder = map.computeIfAbsent(key, k -> new Adder());
adder.like.addAndGet(like);//利用AtomicInteger的原子特性，保证了线程安全
adder.comment.addAndGet(comment);
}
static class Adder{
AtomicInteger like = new AtomicInteger();
AtomicInteger comment = new AtomicInteger();
}
}方案点评：除了没解决回收问题，简单高效
方案三
因为调用save的线程没有并发情况，阻塞也没关系，经分析可巧妙地使用读写锁，同时又不让add方法进入阻塞
/**
* 巧妙地利用读写锁，及save方法可阻塞的特点，实现add操作无阻塞
*/
public class ReadWriteLockCounter implements Counter {
private volatile MapWithLock mapWithLock = new MapWithLock();
@Override
public void save(Saver saver) {
MapWithLock preMapWithLock = mapWithLock;
mapWithLock = new MapWithLock();
//不会一直阻塞，因为mapWithLock已被替换，新的add调用会拿到新的mapWithLock
preMapWithLock.lock.writeLock().lock();
preMapWithLock.map.forEach((key,value)->{
//value已经废弃，故无需value.like.getAndSet(0)
saver.save(key, value.like.get(), value.comment.get());
});
//不能释放该锁，否则add方法中，对被替换掉的MapWithLock.lock执行tryLock会成功
//也许，这是你第一次见到的不需要且不允许释放的锁:)
}
@Override
public void add(String key, int like, int comment) {
MapWithLock mapWithLock;
//如果通过tryLock获取锁失败，则表示该mapWithLock已经被废弃了（因为只有废弃了的MapWithLock才会加写锁），故重新获取最新的mapWithLock
while(!(mapWithLock = this.mapWithLock).lock.readLock().tryLock());
try{
Adder adder = mapWithLock.map.computeIfAbsent(key, k -> new Adder());
adder.like.getAndAdd(like);
adder.comment.getAndAdd(comment);
}finally {
mapWithLock.lock.readLock().unlock();
}
}
static class Adder{
private AtomicInteger like = new AtomicInteger();
private AtomicInteger comment = new AtomicInteger();
}
static class MapWithLock{
private ConcurrentHashMap<String, Adder> map = new ConcurrentHashMap<>();
private ReadWriteLock lock = new ReentrantReadWriteLock();
}
}方案点评：减少了锁的粒度，同时add线程可以相互兼容，大幅提升了并发能力，save线程虽会阻塞，但结合其定时执行的特点，并不受影响，且即使极端情况也不会一直阻塞
方案四
使用一个原子的state来替换LockCounter中的ReadWriteLock(因为只使用到了它的部分特性)，实现wait-free，获得更高性能
/**
* ReadWriteLockCounter的改进版，去掉ReadWriteLock，结合当前场景，实现一个wait-free的简易读写锁<br/>
*/
public class CustomLockCounter implements Counter {
private volatile MapWithState mapWithState = new MapWithState();
@Override
public void save(Saver saver) {
MapWithState preMapWithState = mapWithState;
mapWithState = new MapWithState();
//compareAndSet失败则表示该MapWithState正在被使用，等其使用完，它不会一直失败，因为mapWithState已经被替换
while(!preMapWithState.state.compareAndSet(0,Integer.MIN_VALUE)){
Thread.yield();
}
preMapWithState.map.forEach((key, value)->{
//value已经废弃，故无需value.like.getAndSet(0)
saver.save(key, value.like.get(), value.comment.get());
});
}
@Override
public void add(String key, int like, int comment) {
MapWithState mapWithState;//add的并发，不可能将Integer.MIN_VALUE自增成正数(设置为Integer.MIN_VALUE时，该MapWithState已经被废弃了)
while((mapWithState = this.mapWithState).state.getAndIncrement()<0);
try{
Adder adder = mapWithState.map.computeIfAbsent(key, k -> new Adder());
adder.like.getAndAdd(like);
adder.comment.getAndAdd(comment);
}finally {
mapWithState.state.getAndDecrement();
}
}
static class Adder{
private AtomicInteger like = new AtomicInteger();
private AtomicInteger comment = new AtomicInteger();
}
static class MapWithState {
private ConcurrentHashMap<String, Adder> map = new ConcurrentHashMap<>();
private AtomicInteger state = new AtomicInteger();
}
}方案点评：保留了前一方案ReadWriteLockCounter的优点，同时结合场景的特点做了些优化，本质就是将CAS失败重试循环替换成了一条fetch-and-add指令，如果不是因为save是低频执行，本方案可能是最高效的了(暂且忽略ConcurrentHashMap等其它可能的优化空间)
方案五
先假定不会发生竞争，然后检测竞争情况，如果发生竞争，则补偿
/**
* 乐观地假定不会发生竞争，如果发生了，则尝试进行补偿
*/
public class CompensationCounter implements Counter {
private ConcurrentHashMap<String, Adder> map = new ConcurrentHashMap<>();
@Override
public void save(Saver saver) {
for(Iterator<Map.Entry<String, Adder>> it = map.entrySet().iterator(); it.hasNext();){
Map.Entry<String, Adder> entry = it.next();
it.remove();
entry.getValue().discarded = true;
saver.save(entry.getKey(), entry.getValue().like.getAndSet(0), entry.getValue().comment.getAndSet(0));//需将计数器置0，此处存在竞争
}
}
@Override
public void add(String key, int like, int comment) {
Adder adder = map.computeIfAbsent(key, k -> new Adder());
adder.like.addAndGet(like);
adder.comment.addAndGet(comment);
if(adder.discarded){//如果数量加在了废弃的Adder上面，则执行补偿逻辑
int likeTemp = adder.like.getAndSet(0);
int commentTemp = adder.comment.getAndSet(0);
//即使此后又有线程在计数器上计数了也无妨
if(likeTemp != 0 || commentTemp != 0){
add(key, likeTemp, commentTemp);//补偿
}//也可能已经被其它线程取走了，但并不影响业务正确性
}
}
static class Adder{
AtomicInteger like = new AtomicInteger();
AtomicInteger comment = new AtomicInteger();
volatile boolean discarded = false;//只有保存线程会将它改为true，故使用volatile便能保证线程安全
}
}方案点评：跟乐观锁的思路类似，在竞争激烈的情况下，一般不会有最优性能，但此处因为save方法是低频执行的且自身无并发，add方法才有高并发，故失败补偿其实很少真正被执行，这也是为什么测试结果中本方案性能最优的原因
性能测试
最终我们来测试一下各方案的性能，因为我们抽象出了一个统一的接口，故测试也较为容易
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
public class CounterTester {
private static final int THREAD_SIZE = 6;//add方法的并发线程数
private static final int ADD_SIZE = 5000000;//测试规模
private static final int KEYS_SIZE = 128*1024;
public static void main(String[] args) throws InterruptedException {
Counter[] counters = new Counter[]{new SynchronizedCounter(), new IncompleteCounter(), new ReadWriteLockCounter(), new CustomLockCounter(), new CompensationCounter()};
String[] keys = new String[KEYS_SIZE];
Random random = new Random();
for (int i = 0; i < keys.length; i++) {
keys[i]=String.valueOf(random.nextInt(KEYS_SIZE*1024));
}
for (Counter counter : counters) {
AtomicInteger totalLike = new AtomicInteger();
AtomicInteger totalComment = new AtomicInteger();
AtomicInteger savedTotalLike = new AtomicInteger();
AtomicInteger savedTotalComment = new AtomicInteger();
Counter.Saver saver = (key, like, comment) -> {
savedTotalLike.addAndGet(like);//模拟被持久化到数据库，记录数量以便后续校验正确性
savedTotalComment.addAndGet(comment);//同上
};
CountDownLatch latch = new CountDownLatch(THREAD_SIZE);
long start = System.currentTimeMillis();
for (int i = 0; i < THREAD_SIZE; i++) {
new Thread(()->{
Random r = new Random();
int like, comment;
for (int j = 0; j < ADD_SIZE; j++) {
like = 2;
comment = 4;
counter.add(keys[r.nextInt(KEYS_SIZE)], like, comment);
totalLike.addAndGet(like);
totalComment.addAndGet(comment);
}
latch.countDown();
}).start();
}
Thread saveThread = new Thread(()->{
while(latch.getCount() != 0){
try {
Thread.sleep(100);//模拟100毫秒执行一次持久化
} catch (InterruptedException e) {}
counter.save(saver);
}
counter.save(saver);
});
saveThread.start();
latch.await();
System.out.println(counter.getClass().getSimpleName() +" cost:\t"+(System.currentTimeMillis() - start));
saveThread.join();
boolean error = savedTotalLike.get() != totalLike.get() || savedTotalComment.get() != totalComment.get();
(error?System.err:System.out).println("saved:\tlike="+savedTotalLike.get()+"\tcomment="+savedTotalComment.get());
(error?System.err:System.out).println("added:\tlike="+totalLike.get()+"\tcomment="+totalComment.get()+"\n");
}
}
}在jdk11(jdk8也基本一致)下的测试结果如下：
注：方案二的IncompleteCounter并未完成回收，仅作对比
SynchronizedCounter cost: 12377
saved: like=60000000 comment=120000000
added: like=60000000 comment=120000000
IncompleteCounter cost: 2560
saved: like=60000000 comment=120000000
added: like=60000000 comment=120000000
ReadWriteLockCounter cost: 7902
saved: like=60000000 comment=120000000
added: like=60000000 comment=120000000
CustomLockCounter cost: 3541
saved: like=60000000 comment=120000000
added: like=60000000 comment=120000000
CompensationCounter cost: 2093
saved: like=60000000 comment=120000000
added: like=60000000 comment=120000000
小结
非阻塞同步算法一般不需要我们去设计，直接使用现有的工具便可，但如果真想通过它进一步去压榨性能，应细心分析各线程穿插执行的情况，同时结合业务场景来考虑(也许在A场景不允许的情况，在B场景是允许的)
原创文章，转载请注明： 转载自并发编程网 – ifeve.com本文链接地址: 非阻塞同步算法实战（四）- 计数器定时持久化
(10 votes, average: 4.30 out of 5)
Loading...
第三个方案中的save方法可以释放锁啊
文中的注释分析了此情况，如果释放锁的话，在add方法中，对被替换掉的MapWithLock.lock执行tryLock会成功，然后拿到了一个废弃的计数器，并在上面计数，导致计数丢失。
可以将释放锁的逻辑加上，然后多测试几遍(概率比较小，我跑了20多遍才出现)，会出现计数丢失的情况（打印出的saved和added不相同）