Java 多线程 - Eratosthenes 筛法的基本并行示例

Question

我想创建一个简单的并行 Sieve of Erastosthenes Java 程序，它至少比我在下面发布的串行版本更有效。

  public void runEratosthenesSieve(int upperBound) {
      int upperBoundSquareRoot = (int) Math.sqrt(upperBound);
      boolean[] isComposite = new boolean[upperBound + 1];
      for (int m = 2; m <= upperBoundSquareRoot; m++) {
            if (!isComposite[m]) {
                  System.out.print(m + " ");
               int threads=4;
               for (int n=1; n<=threads; n++) {
                  int job;
                  if (n==1) {job = m * m;} else {job = (n-1)*upperBound/threads;}
                  int upToJob = n*upperBound/threads; 
                   for (int k = job;  k <= upToJob; k += m)
                  {
                       isComposite[k] = true;
                  }               
               } 
            }
      }
      for (int m = upperBoundSquareRoot; m <= upperBound; m++)
            if (!isComposite[m])
                  System.out.print(m + " ");
  }

我创建了一个循环来为 4 个线程分配工作。虽然我不知道如何从中制作实际的线程代码。如何发送变量并启动 4 个线程，每个线程都有部分工作。

Answer 1

我可以提出以下解决方案:有 4 个工作线程和 1 个主线程。工作线程从队列中获取作业。工作基本上是 3 个数字:从、到、步骤。 Master 的意思是 while 必须等到所有线程都完成。完成后，它会搜索下一个质数并创建 4 个工作岗位。可以使用 Semaphore 实现 master 和 worker 之间的同步: master 尝试获得 4 个 permit，而每个 worker 完成后释放 1 个 permit。

public class Sieve {

    // Number of workers. Make it static for simplicity.
    private static final int THREADS = 4;

    // array must be shared between master and workers threads so make it class property.
    private boolean[] isComposite;
    // Create blocking queue with size equal to number of workers.
    private BlockingQueue<Job> jobs = new ArrayBlockingQueue<Job>(THREADS);
    private Semaphore semaphore = new Semaphore(0);
    // Create executor service in order to reuse worker threads. 
    // we can use just new Thread(new Worker()).start(). But using thread pools more effective.
    private ExecutorService executor = Executors.newFixedThreadPool(THREADS);

    public void runEratosthenesSieve(int upperBound) {
        int upperBoundSquareRoot = (int) Math.sqrt(upperBound);
        isComposite = new boolean[upperBound + 1];

        // Start workers.
        for (int i = 0; i < THREADS; i++) {
            executor.submit(new Worker());
        }
        for (int m = 2; m <= upperBoundSquareRoot; m++) {
            if (!isComposite[m]) {
                System.out.print(m + " ");
                for (int n=1; n<= THREADS; n++) {
                    int from;
                    if (n == 1) {
                        from = m * m;
                    } else {
                        from = (n-1)*upperBound/THREADS;
                    }
                    Job job = new Job(from, n*upperBound/threads, m);
                    // Submit job to queue. We don't care which worker gets the job.
                    // Important only that only 1 worker get the job. But BlockingQueue does all synchronization for us.
                    jobs.put(job);
                }
                // Wait until all jobs are done.
                semaphore.acquire(THREADS);
            }
        }
        for (int i = 0; i < n; i++) {
            // put null to shutdown workers.
            jobs.put(null);
        }
        for (int m = upperBoundSquareRoot; m <= upperBound; m++) {
            if (!isComposite[m]) {
                System.out.print(m + " ");
            }
        }
    }

    private class Job {
        public int from, to, step;

        public Job(int from, int to, int step) {
            this.from = from;
            this.to = to;
            this.step = step;
        }
    }

    private Worker implements Runnable {
        while (true) {
            Job job = jobs.take();
            // null means workers must shutdown
            if (job == null) {
                return;
            }
            for (int i = job.from; i <= job.to; i += job.step) {
                isComposite[i] = true;
            }
            // Notify master thread that a job was done.
            semaphore.release();
        }
    }
}