Java BlockingQueue não tem uma espiada em bloqueio?

Question

Eu tenho uma fila de objetos de bloqueio.

Eu quero escrever um tópico que bloqueie até que haja um objeto na fila. Semelhante à funcionalidade fornecida pelo BlockingQueue.take ().

No entanto, como não sei se poderei processar o objeto com sucesso, quero apenas espiar () e não remover o objeto. Quero remover o objeto apenas se puder processá -lo com sucesso.

Então, eu gostaria de uma função bloqueadora (). Atualmente, Peek () retorna se a fila estiver vazia de acordo com os javadocs.

Estou esquecendo de algo? Existe outra maneira de alcançar essa funcionalidade?

EDITAR:

Alguma idéia de se eu apenas usasse uma fila segura para tópicos e espiei e dormi?

public void run() {
    while (!exit) {
        while (queue.size() != 0) {
            Object o =  queue.peek();
            if (o != null) {
                if (consume(o) == true) {
                    queue.remove();
                } else {
                    Thread.sleep(10000); //need to backoff (60s) and try again
                }
            }
        }
        Thread.sleep(1000); //wait 1s for object on queue
    }
}

Observe que eu tenho apenas um thread do consumidor e um thread (separado) do produtor. Eu acho que isso não é tão eficiente quanto usar um BlockingQueue ... quaisquer comentários apreciados.

Answer 1

Você poderia usar um LinkedBlockingDeque e remova fisicamente o item da fila (usando takeLast()) mas substitua -o novamente em o fim da fila Se o processamento falhar usando putLast(E e). Enquanto isso, seus "produtores" adicionariam elementos ao frente da fila usando putFirst(E e).

Você sempre pode encapsular esse comportamento dentro do seu próprio Queue implementação e fornecer um blockingPeek() método que executa takeLast() Seguido por putLast() nos bastidores sobre o subjacente LinkedBlockingDeque. Portanto, da perspectiva do cliente de chamada, o elemento nunca é removido da sua fila.

Answer 2

No entanto, como não sei se poderei processar o objeto com sucesso, quero apenas espiar () e não remover o objeto. Quero remover o objeto apenas se puder processá -lo com sucesso.

Em geral, não é seguro para fios. E se, depois de você peek() e determine que o objeto pode ser processado com sucesso, mas antes de você take() Para remover e processar, outro encadeamento leva esse objeto?

Answer 3

A única coisa que eu estou ciente disso é BlockingBuffer dentro Coleções do Apache Commons:

Se o Get Or Remover for chamado em um buffer vazio, o Thread Whread aguarda a notificação de que uma operação ADD ou ADDALL foi concluída.

get() é equivalente a peek(), e a Buffer pode ser feito para agir como BlockingQueue decorando a Ilimitadoffobuffer com um BlockingBuffer

Answer 4

Você também poderia apenas adicionar uma fila do ouvinte de eventos à sua fila de bloqueio; então, quando algo é adicionado à fila (bloqueando), enviar um evento para seus ouvintes? Você pode ter seu bloco de threads até que o método de desempenho acionário fosse chamado.

Answer 5

A resposta rápida é: não há realmente uma maneira de ter uma espiada, o bar implementando uma fila de bloqueio com uma espiada bloqueadora ().

Estou esquecendo de algo?

Peek () pode ser problemático com a simultaneidade -

• Se você não puder processar sua mensagem de Peek () - ela será deixada na fila, a menos que você tenha vários consumidores.
• Quem vai tirar esse objeto da fila se você não puder processá -lo?
• Se você possui vários consumidores, obtém uma condição de corrida entre você () '' ing e outro thread também processando itens, resultando em processamento duplicado ou pior.

Parece que você pode estar melhor removendo o item e processá -lo usando umPadrão de cadeia de responsabilidade

EDIT: Re: Seu último exemplo: se você tiver apenas 1 consumidor, nunca se livrará do objeto na fila - a menos que seja atualizado nesse meio tempo - nesse caso, é melhor ter muito cuidado com a segurança do tópico e Provavelmente não deveria ter colocado o item na fila de qualquer maneira.

Answer 6

Parece que o próprio BlockingQueue não tem a funcionalidade que você está especificando.

Eu posso tentar reestruturar um pouco o problema um pouco: o que você faria com objetos que não pode "processar corretamente"? Se você está deixando -os na fila, terá que retirá -los em algum momento e lidar com eles. Eu recomendaria descobrir como processá -los (geralmente, se uma fila.get () der algum tipo de valor inválido ou ruim, você provavelmente está bem para soltar no chão) ou escolhendo uma estrutura de dados diferente de Um FIFO.

Answer 7

Não é uma resposta em si, mas: JDK-6653412 Reivindicações Este não é um caso de uso válido.

Answer 8

A solução 'mais simples'

Não processe o próximo elemento até o anterior O elemento é processado com sucesso.

public void run() {

Object lastSuccessfullyProcessedElement = null;

    while (!exit) {
        Object obj =  lastSuccessfullyProcessedElement == null ? queue.take() : lastSuccessfullyProcessedElement; // blocking

        boolean successful = process(obj);

        if(!successful) {
            lastSuccessfullyProcessedElement = obj;
        } else {
            lastSuccessfullyProcessedElement = null;
        }
    }
}

1. Chamando peek() e verificar se o valor for nulo não for eficiente na CPU.

Vi o uso da CPU indo para 10% no meu sistema quando a fila estiver vazia para o programa a seguir.

while (true) {
   Object o = queue.peek();
   if(o == null) continue;
   // omitted for the sake of brevity
}

2. Adicionando sleep() adiciona lentidão.

3. Adicionando -o de volta à fila usando putLast vai perturbar a ordem. Além disso, é uma operação de bloqueio que requer bloqueios.