【问题标题】:Java Blocking List ImplementationJava 阻塞列表实现
【发布时间】:2015-07-05 13:06:58
【问题描述】:

我在 SO 和 Google 上搜索了这个问题的答案,但到目前为止找不到合适的解决方案。

我目前正在处理图形路由问题中的 LayerManager。管理器负责提供和重置一组固定的层。

我想用一个阻塞列表来实现消费者-生产者模式,这样只要没有可用的空闲层,传入的路由请求就会被阻塞。到目前为止,我只找到了blocking queue,但由于我们不需要 FIFO、LIFO,但随机访问队列并没有真正起作用。更准确地说,这样的事情应该是可能的:

/* this should be blocking until a layer becomes available */
public Layer getLayer(){ 

    for ( Layer layer : layers ) {
        if ( layer.isUnused() && layer.matches(request) )
            return layers.pop(layer);
    }
}

有什么办法可以做到吗?

【问题讨论】:

  • java.util.concurrent.PriorityBlockingQueue 和你自己的比较器怎么样?
  • 谢谢。好吧,在我看来, layer 并没有真正的可比性。它们只匹配给定的请求。
  • @Daniel,我知道这是一个老问题,但是您实施了什么解决方案来解决这个问题?

标签: java list blocking


【解决方案1】:

您要查找的内容称为“信号量”。

  1. 创建信号量类
  2. 将其作为字段添加到图层类

示例

 public class Semaphore 
{
    private boolean signal = false;

    public synchronized boolean take() 
    {
       if(this.signal==true)
            return false;  //already in use
       this.signal = true;
       this.notify();
       return true;
    }

     public synchronized void release() throws InterruptedException
     {
        while(!this.signal) wait();
        this.signal = false;
     }


     public boolean isUnused()
     {
         return !signal ;
     }

}


//2.
class Layer
{
    Semaphore sem =null;

    /*your code*/
     /*sem = new Semaphore(); in constructors*/
    public boolean take()
    {
        return this.sem.take();
    }

    public void release()
    {
        this.sem.release();
    }

    public Layer getLayer()
    { 

        for ( Layer layer : layers ) 
        {
         if ( layer.matches(request) && layer.take())
             return layer;
        }

         return null;
    }
}


同步方法处理访问并发

3. 循环 getLayer 直到

Layer l=null;
while(l==null)
{
    l= getlayer();
    Thread.sleep(100); //set time
}
 // continue
 // do not forget to release the layer when you are done

【讨论】:

  • 这就是我现在所做的,我不喜欢这种通知/等待模式。它比阻塞更容易出错
【解决方案2】:

尝试使用Map<String, BlockingQueue<Layer>>。这个想法是在BlockingQueue 中保留免费层。每个请求都有自己的队列。

public class LayerQueue {

    Map<String, BlockingQueue<Layer>> freeLayers = Collections.synchronizedMap(new HashMap<String, BlockingQueue<Layer>>());

    public LayerQueue() {
        //init QUEUEs
        freeLayers.put("request-1", new ArrayBlockingQueue<Layer>(1)); // one to one...
        freeLayers.put("request-2", new ArrayBlockingQueue<Layer>(1));
        [...] 
    }

    public void addUnusedLayer(Layer layer, String request) {
        BlockingQueue<Layer> freeLayersForRequest = freeLayers.get(request);
        freeLayersForRequest.add(layer);
    }

    public Layer getLayer(String request) {

        BlockingQueue<Layer> freeLayersForRequest = freeLayers.get(request);

        try {
            return freeLayersForRequest.take(); // blocks until a layer becomes available
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }
}

【讨论】:

  • 1 每个请求有一层,地图也不是固定大小的
  • @DanielGerber 为什么它应该是一个问题?
  • 因为 1 层可能需要高达 500mb 的 RAM。
  • @DanielGerber 即使它需要 4GB,它仍然可以工作。您正在使用参考文献。
【解决方案3】:

我不太确定我是否正确理解了您的需求,但您可以使用阻塞队列并将结果放入列表中。如果在列表中未找到合适的层,则调用 wait() 并再次检查新项目何时从队列中添加到列表中。这听起来好像它可以在概念上工作,即使下面的代码没有正确(我很确定这不是完全同步)

public class PredicateBlockingQueue<Product> {

private final List<Product> products = new LinkedList<Product>();
private final BlockingQueue<Product> queue;
private final Thread consumer;

public PredicateBlockingQueue(int capacity) {
    queue = new ArrayBlockingQueue<Product>(capacity);

    consumer = new Thread() {
        @Override
        public void run() {
            while(!Thread.interrupted()) {
                try {
                    products.add(queue.take());
                    synchronized(queue) {
                        queue.notifyAll();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    };

    consumer.start();
}

public void put(Product product) throws InterruptedException {
    queue.put(product);
}

public Product take(Predicate<Product> predicate) throws InterruptedException {
    Product product;
    while((product=find(predicate))==null) {
        synchronized(queue) {
            queue.wait();
        }
    }
    return product;
}

private synchronized Product find(Predicate<Product> predicate) {
    Iterator<Product> it = products.iterator();
    while(it.hasNext()) {
        Product product = it.next();
        if(predicate.test(product)) {
            it.remove();
            return product;
        }
    }
    return null;
}

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2013-12-05
    • 1970-01-01
    • 1970-01-01
    • 2016-08-26
    • 1970-01-01
    • 1970-01-01
    • 2022-11-19
    相关资源
    最近更新 更多