如何手动撸一个能够满足下列并发的队列？

Question

在做puv统计时碰到的一个问题，用户请求过来会记录为一个pv，记录到redis中，但由于pv量太大会给redis造成过大压力，所以做个缓存，当pv满10条了发一次。用了一个队列ArrayList实现，但队列的插入、删除在并发条件下不可行，所以在方法上加了synchronized:

static ArrayList<Object> pvList = new ArrayList<Object>();
public synchronized void countPv(...){
    //........生成一个PV对象
    pvList.add(PV)
    if(pvList.size()>10){
        //前10个加到redis
        addRedis(pvList.subList(0,10))
    }
    //删除10个
    pvList.subList(0,10).clear();
}

但在压力测试中，如果已满负荷的连续压测，发现会丢掉一些pv，可能是synchronized造成的堵塞导致，如何更好的实现这个需求呢？

每10个请求发一次，而后删除，同时可以满足不断累加

1. synchronized放到函数里面估计提升并不大，毕竟add，size，clear方法都必须满足同步需求
2. ConcurrnetLinkedQueue, Concurrent...Array删除、查找的开销都非常大，而且貌似没法用于这种场合

p.s. 您能留段伪码就最好了

Answer 1

public synchronized void countPv(...){
    //........生成一个PV对象
   
    pvList.add(PV)
    if(pvList.size() > 10){
        addRedis(pvList.subList(0,10))
    }

    pvList.subList(0,10).clear();
}

pvList.subList(0,10).clear(); 这句代码不应该写到 if 里面吗？
既然 pvList 是 static 的（类变量），那么 countPV 也应该是 static 的才对，这样 synchronize 使用的锁才会是 class，而不是对象。

Answer 2

不知道你说的满负荷压测时丢pv指的是什么,有没有记录这种情况下,服务器返回的状态码是什么?是因为超时么?
如果是因为压力过大超出服务器载能力,换用或其他数据结构也不见得好到哪去, 如果想解锁synchronized, 可能试试多例模式. 每个服务线程创建自己私有的缓存.

Answer 3

不要用synchronized
ConcurrnetLinkedQueue 用法不对 subList(0,10)？？？ 开个线程从队尾取 取了10个就 处理
你这10个扔redis搞什么意思 （还不如直接扔redis）
可以扔消息队列里面 然后异步的处理

贴一段代码
https://www.cnblogs.com/linji...
运行结果：
costtime 2360ms

改用while (queue.size()>0)后
运行结果：
cost time 46422ms

结果居然相差那么大，看了下ConcurrentLinkedQueue的API原来.size()是要遍历一遍集合的，难怪那么慢，所以尽量要避免用size而改用isEmpty().

总结了下， 在单位缺乏性能测试下，对自己的编程要求更加要严格，特别是在生产环境下更是要小心谨慎。

Answer 4

java线程池，怎么样

ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
    final int index = i;
    singleThreadExecutor.execute(new Runnable() {
 
        @Override
        public void run() {
            try {
                // TODO 干你想干的
                addRedis(pvList.subList(0,10))
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    });
}

Java(Android)线程池

Answer 5

https://stackoverflow.com/que...

Answer 6

@fengdui 有一句评论说得好，Redis操作要搬出去。

方法一：

// 把10个元素return出去让外面的调用者去调Redis，别占用同步块的时间
public synchronized List<Object> countPv(...) {
  ...
}

方法二：开个线程池去异步发Redis，但是机器重启会丢失来不及发送的数据

if (pvList.size() > 10) {
        //前10个打包成任务扔给线程池
        senderExecutor.execute(new SendTask(new ArrayList<>(pvList.subList(0,10)))); //当场复制了subList
        pvList.subList(0,10).clear(); //这行要移到这里，这可能就是你丢数据的原因
}

Answer 7

pipeline ?