标签: blockingqueue

Object当一些BlockingQueue人有一个项目给我时,我需要异步通知.

我已经在Javadoc和网络上搜索了一个预先制定的解决方案,然后我最终得到了一个(也许是天真的)我的解决方案,这里是:

interface QueueWaiterListener<T> {
    public void itemAvailable(T item, Object cookie);
}

和

class QueueWaiter<T> extends Thread {

    protected final BlockingQueue<T> queue;
    protected final QueueWaiterListener<T> listener;
    protected final Object cookie;

    public QueueWaiter(BlockingQueue<T> queue, QueueWaiterListener<T> listener, Object cookie) {
        this.queue = queue;
        this.listener = listener;
        this.cookie = cookie;
    }

    public QueueWaiter(BlockingQueue<T> queue, QueueWaiterListener<T> listener) {
        this.queue = queue;
        this.listener = listener;
        this.cookie = null;
    }

    @Override
    public void run() {
        while (!isInterrupted()) {
            try {
                T item = …

(请在最后阅读更新3)我正在开发一个应用程序,它不断与设备的传感器一起工作,Accelerometer与Magnetic传感器一起工作以检索设备的方向(这里提到的目的).换句话说,我的应用程序需要知道设备的实时方位(然而,这是永远不可能的,所以尽可能快地代替,但真正以最快的速度可能!).正如Reto Meier在专业Android 4应用程序开发中所提到的:

加速度计每秒可以更新数百次......

我不能丢失传感器报告的任何数据,我也想对这些数据进行耗时的操作(检索方向,然后进行计算......).我决定使用LinkedBlockingQueue以下方法解决我的问题:

    public void startSensors() {
            LinkedBlockingQueue<float[][]> array=new LinkedBlockingQueue();
    sensorListenerForOrientation = new SensorEventListener() {

        @Override
        public void onSensorChanged(SensorEvent event) {
            if (event.sensor.getType() == Sensor.TYPE_ACCELEROMETER)
                aValues = (event.values.clone());
            else if (event.sensor.getType() == Sensor.TYPE_MAGNETIC_FIELD)
                mValues = (event.values.clone());
            if (aValues != null && mValues != null) {
                try {
                    array.put(new float[][] { aValues, mValues });
                } catch (InterruptedException e) {
                }
            }
        }

        @Override
        public void …

我正在努力实现我的处理管道的最佳方法.

我的制作人将作品提供给BlockingQueue.在消费者方面,我轮询队列,包装我在Runnable任务中获得的内容,并将其提交给ExecutorService.

while (!isStopping())
{
    String work = workQueue.poll(1000L, TimeUnit.MILLISECONDS);
    if (work == null)
    {
        break;
    }
    executorService.execute(new Worker(work));   // needs to block if no threads!
}

这不理想; 当然,ExecutorService有自己的队列,所以真正发生的事情是我总是完全耗尽我的工作队列并填充任务队列,随着任务的完成,队列会慢慢排空.

我意识到我可以在生产者端排队任务,但我真的不愿意这样做 - 我喜欢我的工作队列的间接/隔离是愚蠢的字符串; 它真的不是生产者的任何事情会发生在他们身上.迫使生产者对Runnable或Callable进行排队会破坏抽象,恕我直言.

但我确实希望共享工作队列代表当前的处理状态.如果消费者没有跟上,我希望能够阻止生产者.

我喜欢使用Executors,但我觉得我正在与他们的设计作斗争.我可以部分喝Kool-ade,还是我必须吞下它？我是否正在抵制排队任务？(我怀疑我可以设置ThreadPoolExecutor来使用1任务队列并覆盖它的执行方法来阻止而不是拒绝队列满,但这感觉很糟糕.)

建议？

任何人都可以解释为什么有人应该使用Android Looper功能来创建一个"管道线程",而不是制作一个从BlockingQueue中提取任务的普通线程？从表面上看,似乎有两种方法可以做同样的事情.

我正在尝试使用URLConnection下载pdf文件.这是我设置连接对象的方法.

URL serverUrl = new URL(url);
urlConnection = (HttpURLConnection) serverUrl.openConnection();
urlConnection.setDoInput(true);
urlConnection.setRequestMethod("GET");
urlConnection.setRequestProperty("Content-Type", "application/pdf");
urlConnection.setRequestProperty("ENCTYPE", "multipart/form-data");
String contentLength = urlConnection.getHeaderField("Content-Length");

我从连接对象获得了输入流.

bufferedInputStream = new BufferedInputStream(urlConnection.getInputStream());

并输出流来写入文件内容.

File dir = new File(context.getFilesDir(), mFolder);
if(!dir.exists()) dir.mkdir();
final File f = new File(dir, String.valueOf(documentName));
f.createNewFile();
final BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(new FileOutputStream(f, true)); //true for appendMode

创建BlockingQueue,以便执行读写操作的线程可以访问队列.

final BlockingQueue<ByteArrayWrapper> blockingQueue = new ArrayBlockingQueue<ByteArrayWrapper>(MAX_VALUE,true);
final byte[] dataBuffer = new byte[MAX_VALUE];

现在创建了从InputStream读取数据的线程.

Thread readerThread = new Thread(new Runnable() {
       @Override
       public void run() {
         try …

考虑一个BlockingQueue和几个线程等待poll(long, TimeUnit)(可能还在上take()).

现在队列是空的,并且希望通知等待的线程他们可以停止等待.预期的行为是null返回或声明InterruptedException抛出.

Object.notify()LinkedBlockingQueue因为线程正在等待内部锁定而无法工作.

有任何直截了当的方式吗？

我正在浏览ArrayBlockingQueue和LinkedBlockingQueue的源代码.LinkedBlockingQueue有一个putLock和一个takeLock分别用于插入和删除,但ArrayBlockingQueue只使用1个锁.我相信LinkedBlockingQueue是基于简单,快速,实用的非阻塞和阻塞并发队列算法中描述的设计实现的.在本文中,他们提到他们保留一个虚拟节点,以便入队者永远不必访问头部,并且出队员永远不必访问尾部,这避免了死锁情况.我想知道为什么ArrayBlockingQueue没有借用相同的想法而是使用2个锁.

所以我在生产者/消费者类型应用程序中使用固定大小的BlockingQueue [ArrayBlockingQueue],但我希望用户能够动态更改队列大小.问题是没有BlockingQueue实现允许在创建后更改容量.以前有人见过这个吗？有任何想法吗？

在C#中,我想知道是否可以等到后台线程清除BlockingCollection,如果时间太长则超时.

我现在拥有的临时代码让我觉得有些不雅(因为什么时候使用它是好的做法Thread.Sleep？):

while (_blockingCollection.Count > 0 || !_blockingCollection.IsAddingCompleted)
{
    Thread.Sleep(TimeSpan.FromMilliseconds(20));
    // [extra code to break if it takes too long]
}

我正在阅读java 8的实际操作,作者引用此链接:http://mail.openjdk.java.net/pipermail/lambda-dev/2013-November/011516.html

并编写自己的流forker,看起来像这样:

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class Main {

    public static void main(String... args) {
        List<Person> people = Arrays.asList(new Person(23, "Paul"), new Person(24, "Nastya"), new Person(30, "Unknown"));
        StreamForker<Person> forker = new StreamForker<>(people.stream())
                .fork("All names", s -> s.map(Person::getName).collect(Collectors.joining(", ")))
                .fork("Age stats", s -> s.collect(Collectors.summarizingInt(Person::getAge)))
                .fork("Oldest", s -> s.reduce((p1, p2) -> p1.getAge() > p2.getAge() ? p1 : p2).get());
        Results results = …