java - 共有リストでのマルチスレッド

Question

Web サイトと、それらの Web サイトをクロールするためのコード ブロックを含むリストが存在するシナリオがあります。各スレッドがリストから 5 つ以上の Web サイトを取得して個別にクロールし、別のスレッドによって収集された同じ Web サイトを取得しないように、マルチスレッドの方法を実装することは可能ですか?

List <String> websiteList;

//crawling code block here

Answer 1

BlockingQueueたとえば、関心のあるすべての消費者が共有できるa を使用できます (明確にするために、エラー処理はスキップされていることに注意してください)。

public static void main(String[] args) throws Exception {
    // for test purposes add 10 integers
    final BlockingQueue<Integer> queue = new LinkedBlockingDeque<Integer>();
    for (int i = 0; i < 10; i++) {
        queue.add(i);    // 
    }

    new Thread(new MyRunnable(queue)).start();
    new Thread(new MyRunnable(queue)).start();
    new Thread(new MyRunnable(queue)).start();

}

static class MyRunnable implements Runnable {
    private Queue<Integer> queue;

    MyRunnable(Queue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while(!queue.isEmpty()) {
            Integer data = queue.poll();
            if(data != null) {
                System.out.println(Thread.currentThread().getName() + ": " + data);
            }
        }
    }
}

Queueが空の場合、はThreads終了し、プログラムは終了します。

Answer 2

を使用してLinkedBlockingQueue、すべての websiteList をこのキューに入れ、このキューを各スレッド間で共有できます。これで、すべてのスレッドがこのキューをポーリングします。これは、1 つの要素が 1 つのスレッドだけによってフェッチされるキューであることを確認するブロッキング操作です。

何かのようなもの：

String site;
while((site=queue.poll(timeout, TimeUnit.SECONDS))!=null)
{
//process site
}

Answer 3

次の3 つの解決策のいずれかをお勧めします。

複雑にしないでおく

synchronized(list) {
    // get and remove 5 websites from the list
}

リストタイプを変更できる場合は、使用できます

BlockingQueue

リストの種類を変更できない場合は、

Collections.synchronizedList(list)

Answer 4

DoubleBufferedList を試すことができます。これにより、複数のスレッドからリストとエントリをリストに追加し、複数のスレッドを使用して完全にロックフリーでリストを取得できます。

public class DoubleBufferedList<T> {
  // Atomic reference so I can atomically swap it through.
  // Mark = true means I am adding to it so momentarily unavailable for iteration.
  private AtomicMarkableReference<List<T>> list = new AtomicMarkableReference<>(newList(), false);

  // Factory method to create a new list - may be best to abstract this.
  protected List<T> newList() {
    return new ArrayList<>();
  }

  // Get and replace the current list.
  public List<T> get() {
    // Atomically grab and replace the list with an empty one.
    List<T> empty = newList();
    List<T> it;
    // Replace an unmarked list with an empty one.
    if (!list.compareAndSet(it = list.getReference(), empty, false, false)) {
      // Failed to replace! 
      // It is probably marked as being appended to but may have been replaced by another thread.
      // Return empty and come back again soon.
      return Collections.<T>emptyList();
    }
    // Successfull replaced an unmarked list with an empty list!
    return it;
  }

  // Grab and lock the list in preparation for append.
  private List<T> grab() {
    List<T> it;
    // We cannot fail so spin on get and mark.
    while (!list.compareAndSet(it = list.getReference(), it, false, true)) {
      // Spin on mark - waiting for another grabber to release (which it must).
    }
    return it;
  }

  // Release the list.
  private void release(List<T> it) {
    // Unmark it - should this be a compareAndSet(it, it, true, false)?
    if (!list.attemptMark(it, false)) {
      // Should never fail because once marked it will not be replaced.
      throw new IllegalMonitorStateException("It changed while we were adding to it!");
    }
  }

  // Add an entry to the list.
  public void add(T entry) {
    List<T> it = grab();
    try {
      // Successfully marked! Add my new entry.
      it.add(entry);
    } finally {
      // Always release after a grab.
      release(it);
    }
  }

  // Add many entries to the list.
  public void add(List<T> entries) {
    List<T> it = grab();
    try {
      // Successfully marked! Add my new entries.
      it.addAll(entries);
    } finally {
      // Always release after a grab.
      release(it);
    }
  }

  // Add a number of entries.
  @SafeVarargs
  public final void add(T... entries) {
    // Make a list of them.
    add(Arrays.<T>asList(entries));
  }
}