0

我有一个使用阻塞的生产者/消费者的实现，我像这样启动它们

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();
}

Producer 内部是这样的

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){}
        }
     }
}

但是一段时间后它变得非常慢，无论如何我可以保持它非常快吗？

4

2 回答 2

0

移除 Consumer 中的休眠，对 queue.take() 的调用将阻塞线程，直到 Objekt 可用，因此无需休眠。

但是，由于它启动速度很快，随着时间的推移会变慢，请尝试优化您的 Producer。尽量在队列中使用更多的元素，优化 Sql.updateTRECEIVE(); 并删除睡眠。

于 2012-03-02T12:29:02.677 回答
0 
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
于 2017-02-25T15:55:46.813 回答