没有任何一个操作ArrayBlockingQueue与其他任何操作并发; 他们总是采取同样的锁.即使对于该size()方法,它也需要锁定.
public int size() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}
虽然执行LinkedBlockingQueue你有两个锁:put和take.并且因为size()它使用AtomicInteger所以不需要锁.
所以我的问题是:为什么这个实现在并发包中 - ArrayBlockingQueue真的是并发的？
问题:有多个线程访问资源.我需要将它们的数量限制为常数MaxThreads.无法进入线程池的线程应该收到错误消息.
解决方案:我开始BlockingCollection<string> pool在下面的算法中使用a ,但是我看到BlockingCollection需要调用CompleteAdding,我不能这样做,因为我总是得到传入的线程(我在下面的示例中硬编码为10用于调试目的),想想Web请求.
public class MyTest {
private const int MaxThreads = 3;
private BlockingCollection<string> pool;
public MyTest() {
pool = new BlockingCollection<string>(MaxThreads);
}
public void Go() {
var addSuccess = this.pool.TryAdd(string.Format("thread ID#{0}", Thread.CurrentThread.ManagedThreadId));
if (!addSuccess) Console.WriteLine(string.Format("thread ID#{0}", Thread.CurrentThread.ManagedThreadId));
Console.WriteLine(string.Format("Adding thread ID#{0}", Thread.CurrentThread.ManagedThreadId));
Console.WriteLine(string.Format("Pool size: {0}", pool.Count));
// simulate work
Thread.Sleep(1000);
Console.WriteLine("Thread ID#{0} " + Thread.CurrentThread.ManagedThreadId + " is done doing work.");
string val;
var takeSuccess = this.pool.TryTake(out val);
if (!takeSuccess) …我在多线程系统中使用BlockingQueue,其中synchronized块将项添加到列表中.有时它不会将项目添加到列表中,它未命中的项目是随机的.我尝试在代码中添加以下行,然后它从未错过任何项目.
list.forEach(item -> logger.info(" In list "+item));
我觉得这种行为有点奇怪.有人可以帮我弄清楚如何解决这个丢失文件的问题？我不想不必要地遍历整个列表.我错过了什么吗？
我试图让线程池的基础知识非常强大。我了解到它在内部使用阻塞队列来“窃取”任务并将它们运行到池中的给定线程中。这意味着如果我有 10 个任务和 5 个线程，它只能同时运行 5 个任务，直到 1 个任务完全完成。
问题是：为什么不并发？为什么不只是对这 10 个任务进行时间切片？这个实现的原因是什么？
我最近遇到了一些线程相关的问题,消费者需要积分.这是原始的,除了占用大量的cpu不断检查队列之外,它工作正常.这个想法是可以随便调用cuePoint,主线程继续运行.
import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class PointConsumer implements Runnable {
public static final int MAX_QUEUE_SIZE=500;
BlockingQueue<Point> queue;
public PointConsumer (){
this.queue=new ArrayBlockingQueue<Point>(MAX_QUEUE_SIZE);
}
public void cuePoint(Point p){
try{
this.queue.add(p);
}
catch(java.lang.IllegalStateException i){}
}
public void doFirstPoint(){
if(queue.size()!=0){
Point p=queue.poll();
//operation with p that will take a while
}
}
public void run() {
while(true){
doFirstPoint();
}
}
}
我试图通过每次调用cue函数时添加notify()来修复cpu问题,并将doFirstPoint()重新处理为这样的事情:
public void doFirstPoint(){
if(queue.size()!=0){
//operation with p that will take a while
}
else{
try{
wait(); …我最近看到了以下针对BlockingQueue的入队实现(源代码)
public synchronized void enqueue(Object item)
throws InterruptedException {
while(this.queue.size() == this.limit) {
wait();
}
if(this.queue.size() == 0) {
notifyAll();
}
this.queue.add(item);
}
为什么while循环是必要的,可以while替换为if (this.queue.size() == this.limit)
似乎方法enqueue是同步的,因此一次只有1个线程可以在方法体中执行并进行调用wait().一旦线程被通知,它不能继续向前而不再检查this.queue.size() == this.limit条件？
我认为PriorityBlockingQueue按优先顺序排序但结束不是我的预期,谁可以告诉我原因.
public class TestPriorityQueue {
static Random r=new Random(47);
public static void main(String args[]) throws InterruptedException{
final PriorityBlockingQueue q=new PriorityBlockingQueue();
ExecutorService se=Executors.newCachedThreadPool();
//execute producer
se.execute(new Runnable(){
public void run() {
int i=0;
while(true){
q.put(new PriorityEntity(r.nextInt(10),i++));
try {
TimeUnit.MILLISECONDS.sleep(r.nextInt(1000));
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
});
//execute consumer
se.execute(new Runnable(){
public void run() {
while(true){
try {
System.out.println("take== "+q.take()+" left:== ["+q.toString()+"]");
try {
TimeUnit.MILLISECONDS.sleep(r.nextInt(1000));
} catch (InterruptedException e) {
// TODO Auto-generated …我试图了解如何在Linux内核中实现wait_event。ldd3中有一个代码示例，其中使用prepare_to_wait（http://www.makelinux.net/ldd3/chp-6-sect-2）解释了内部实现。
static int scull_getwritespace(struct scull_pipe *dev, struct file *filp)
{
while (spacefree(dev) == 0) {
DEFINE_WAIT(wait);
up(&dev->sem);
if (filp->f_flags & O_NONBLOCK)
return -EAGAIN;
PDEBUG("\"%s\" writing: going to sleep\n",current->comm);
prepare_to_wait(&dev->outq, &wait, TASK_INTERRUPTIBLE);
if (spacefree(dev) == 0) // Why is this check necessary ??
schedule( );
finish_wait(&dev->outq, &wait);
if (signal_pending(current))
return -ERESTARTSYS; /* signal: tell the fs layer to handle it */
if (down_interruptible(&dev->sem))
return -ERESTARTSYS;
}
return 0;
}
在书中，解释如下。
然后是对缓冲区的强制检查；我们必须处理以下情况：在进入while循环（并丢弃了信号灯）之后但在进入等待队列之前，缓冲区中的空间变得可用。如果没有该检查，如果读取器进程能够在那个时候完全清空缓冲区，那么我们可能会错过唯一一次唤醒并永远休眠的唤醒。在满足自己必须睡觉的满足感之后，我们可以拨打时间表了。
我无法理解这一解释。如果if (spacefree(dev) == 0)在调用schedule（）之前未完成睡眠，我们将如何进入不确定的睡眠状态？如果不存在此强制检查，则wakeup（）仍将过程状态重置为TASK_RUNNING并按下一段所述返回计划。
值得再次研究这种情况：如果在if语句中的测试与计划调用之间发生唤醒，会发生什么情况？在这种情况下，一切都很好。唤醒会将过程状态重置为TASK_RUNNING并返回计划-尽管不一定立即进行。只要在流程将自己置于等待队列并更改其状态之后进行测试，事情就可以正常进行。