java - Concurrentlinkedqueue在多线程环境下未命中添加数据

Question

在下面的代码中，在极其罕见的情况下(10 亿次 QueueThread 对象执行中有 3 次)如果 block 和 queue.size 结果为 7999，它会达到下面提到的。可能的原因是什么。

 if(q.size()<batchsize){   
     System.out.println("queue size" +q.size());   
 }

基本上它没有执行 queue.add 语句但执行了线程中的所有其他语句。

代码片段如下。

import java.util.concurrent.ConcurrentLinkedQueue;   
import java.util.concurrent.ExecutorService;   
import java.util.concurrent.Executors;   
import java.util.concurrent.atomic.AtomicInteger;   

public class CLinkQueueTest {   


    public static final int itersize=100000;   
    public static final int batchsize=8000;   
    public static final int poolsize=100;   

    public static void main (String args[]) throws Exception{   
        int j= 0;   
        ExecutorService service = Executors.newFixedThreadPool(poolsize);   
        AtomicInteger counter = new AtomicInteger(poolsize);   
        ConcurrentLinkedQueue<String> q = new ConcurrentLinkedQueue<String>();   
        String s ="abc";   

        while(j<itersize){   
            int k=0;   
            while(k<batchsize){   
                counter.decrementAndGet();   
                service.submit(new QueueThread(counter, q, s));   
                if(counter.get()<=0){   
                    Thread.sleep(5);   
                }   
                k++;   
            }   
            if(j%20 ==0){   
                System.out.println("Iteration no " + j);   
            }   
            while(counter.get() < poolsize){   
                //wait infinitely   
            }   
            if(q.size()<batchsize){   
                System.out.println("queue size" +q.size());   
            }   
            q.clear();   
            j++;               
        }   

        System.out.println("process complete");   
    }   


import java.util.Queue;   
import java.util.concurrent.Callable;   
import java.util.concurrent.ConcurrentLinkedQueue;   
import java.util.concurrent.atomic.AtomicInteger;   

public class QueueThread implements Callable<Boolean> {   

    private AtomicInteger ai;   
    private Queue<String> qu;   
    private String st;   

    public QueueThread(AtomicInteger i, Queue<String> q, String s){   
        ai = i;   
        qu = q;   
        st = s;        
    }   

    @Override  
    public Boolean call() {   
        try{   
            qu.add(st);            
        } catch(Throwable e){   
            e.printStackTrace();   
        }finally{   
            ai.incrementAndGet();   

        }   
        return true;   
    }   

}  

Answer 1

会不会是有一次在队列中注册的条目太少是因为Executor还没有处理完？

很明显，每次调用 QueueThread.call() 时，都会添加队列，并且 AtomicInteger 会递增。我能想到的是一个 call 没有被执行。

也许你可以通过使用类似的东西对系统更友善一些:

        while(counter.get() < poolsize){   
            //wait infinitely   
            Thread.currentThread().sleep(5);
        }   

但这只是我的意见。