使用阻塞队列的生产者-消费者

Question

我使用Blocking实现了生产者/消费者，我像这样启动它们

BlockingQueue<Object> myQueue1 = new LinkedBlockingQueue<Object>();
new Thread(new SmsInProducer(myQueue1, 100)).start();
for (int i = 0; i < 100; i++ ) {
    new Thread(new SmsInConsumer(myQueue1)).start();
}

生产者内部看起来像这个

public class SmsInProducer implements Runnable {
    protected BlockingQueue queue;
    protected int MAX_RECORDS = 5000;
    @SuppressWarnings("rawtypes")
    public SmsInProducer(BlockingQueue theQueue, int maxRecord) {
        this.queue = theQueue;
        this.MAX_RECORDS = maxRecord;
    }
    @SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public void run() {
        // TODO Auto-generated method stub
    while (true) {
            int totalRecords = MAX_RECORDS - queue.size();
            if (totalRecords > 0) {
            List<Map> tList = Sql.updateTRECEIVE();
            if (tList != null && !tList.isEmpty()) {
                queue.addAll(tList);
            }
        }
        // wait for 1 second
    try { Thread.sleep(Parameter.replyWaitTime * 1000); }   catch(Exception e) {}
    }
}

消费者看起来像这个

public class SmsInConsumer implements Runnable {
@SuppressWarnings("rawtypes")
protected BlockingQueue queue;
@SuppressWarnings("rawtypes")
public SmsInConsumer(BlockingQueue theQueue) {
        this.queue = theQueue;
    }
@SuppressWarnings("rawtypes")
@Override
public void run() {
        // TODO Auto-generated method stub
        while (true) {
        try {
        Object obj = queue.take();
                Map map = (Map) obj;
        } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        try { Thread.sleep(Parameter.replyWaitTime * 1000); } catch(Exception e){}
        }
     }
}

但过了一段时间，它变得很慢，有没有办法让我保持它很快？

Answer 1

删除Consumer中的睡眠，对queue.take（）的调用将阻塞线程，直到Objekt可用，因此无需睡眠。

但是，由于它开始得很快，而且随着时间的推移会变慢，请尝试优化您的Producer。尝试在队列中使用更多元素，优化Sql.updateTRECEIVE（）；和取消睡眠。

Answer 2

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
/**
 *
 * @author sakshi
 */
public class BlockingqueueDemo {
    static BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();
    static class Producer extends Thread {
        BlockingQueue queue;
        public Producer(BlockingQueue<Integer> queue) {
            this.queue = queue;
        }
        public void run() {
            for (int i = 0; i < 10; i++) {
                System.out.println("produce" + i);
                try {
                    queue.put(i);
                    Thread.sleep(2000);
               } catch (InterruptedException ex) {
                    Logger.getLogger(BlockingqueueDemo.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    }
    static class Consumer extends Thread {
        BlockingQueue queue;
        public Consumer(BlockingQueue<Integer> queue) {
            this.queue = queue;
        }
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    System.out.println("consume" + queue.take());
                } catch (InterruptedException ex) {
                    Logger.getLogger(BlockingqueueDemo.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    }
    public static void main(String[] args) {
        Producer produce = new Producer(queue);
        Consumer consume = new Consumer(queue);
        produce.start();
        consume.start();
    }
}
output:
produce0
consume0
produce1
consume1
produce2
consume2
produce3
consume3
produce4
consume4
produce5
consume5
produce6
consume6
produce7
consume7
produce8
consume8
produce9
consume9