主线程使用者和其他线程生产者

Question

下面是代码，我面临的问题是recordRead变量告诉线程应该从哪里开始读取记录的起点。但是我如何为每个线程设置不同的值？例如，对于thread1，它应该是0，recordsToRead应该是300，对于thread2，recordsToRead应该是300+300=600，对于最后一个线程，它应该是600以及更高的结束。pagesize=50pagesize、recordRead和recordToRead都是属于主类和主线程的变量。

    ExecutorService service = Executors.newFixedThreadPool(nThreads);
    while(nThreads > 0) {
        nThreads--;
        service.execute(new Runnable() {

            @Override
            public void run() {
                // TODO Auto-generated method stub

                do {
                    int respCode = 0;
                    int RecordsToRead = div;
                    JSONObject jsObj = new JSONObject();
                    jsObj.put("pagesize", pageSize);
                    jsObj.put("start", recordsRead);
                    jsObj.put("searchinternalid", searchInternalId);

                    try {
                        boolean status = req.invoke(jsObj); 
                        respCode = req.getResponseCode();

                    } catch (Exception e) {         
                        req.reset();
                        e.printStackTrace();
                        return true;
                    }
                    JSONObject jsResp = req.getResponseJson();
                    //here jsResp will be added to ArrayBlockingQueue.

                    req.reset();
                }while(!isError && !isMaxLimit && recordsRead < RecordsToRead);

            }

        });
    }

在这个循环之后将是主线程读取队列的代码。如何为所有线程设置recordsRead和recordToread。

以及如何使主线程等待直到至少有一个线程在队列中插入一个对象。

Answer 1

我在你的定义中看到了两个问题。第一个问题是执行并行块计算，第二个问题是创建一个连续的流水线。让我们从第一个问题开始。为了实现具有预定义大小的并行计算，fmpv的最佳选择是使用fork-join框架。不仅是因为性能（工作窃取确实有效），也是因为代码更简单。但是，由于您被限制为3个线程对我来说，它似乎也有效地直接使用线程。简单地说，你想要的东西我可以通过这种方式来实现：

    final int chunkSize = 300;
    //you can also use total amount of job
    //int totalWork = 1000 and chunk size equals totalWork/threadsNumber
    final int threadsNumber = 3;

    Thread[] threads = new Thread[threadsNumber];

    for (int ii = 0; ii < threadsNumber; ii++) {
        final int i = ii;

        threads[ii] = new Thread(() -> {
           //count your variable according the volume
            // for example you can do so
            int chunkStart = i * chunkSize; 
            int chunkEnd = chunkStart + chunkSize;
            for(int j = chunkStart; j < chunkEnd; j++) {
              //object creation with necessary proprs
              //offer to queue here
            }
        });

        threads[ii].start();
    }

    //your code here
    //take here

    for (int ii = 0; ii < threadsNumber; ii++) {
        try {
         //this part is only as example
         //you do not need it                
         //here if you want you can also w8 for completion of all threads
            threads[ii].join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

关于消费的第二个问题。对于这个puprose，您可以使用ConcurrentLinkedBlockingQueue（http://www.jgroups.org/javadoc/org/jgroups/util/ConcurrentLinkedBlockingQueue.html）。在生产者线程中报价，主要采用取用方法。

但老实说，我还是不明白你问题的原因。您想创建连续的管道还是它只是一次性的计算？