Как читать уникальные элементы из массива на поток?

Ответ 1

Ваши детали вопроса неоднозначны - в вашем примере показано, что двум потокам может передаваться тот же Pair, но вы говорите иначе в описании.

Чем труднее достичь, я предлагаю Iterable<Pair<Q,E>>, который будет доставлять Pair по одному на поток до тех пор, пока поставщик не начнет цикл, а затем повторится.

public interface Supplier<T> {
  public int size();

  public T get(int index);

}

public interface PairSupplier<Q, E> extends Supplier<Pair<Q, E>> {
}

public class IterableSupplier<T> implements Iterable<T> {
  // The common supplier to use across all threads.
  final Supplier<T> supplier;
  // The atomic counter.
  final AtomicInteger i = new AtomicInteger();

  public IterableSupplier(Supplier<T> supplier) {
    this.supplier = supplier;
  }

  @Override
  public Iterator<T> iterator() {
    /**
     * You may create a NEW iterator for each thread while they all share supplier
     * and Will therefore distribute each Pair between different threads.
     *
     * You may also share the same iterator across multiple threads.
     *
     * No two threads will get the same pair twice unless the sequence cycles.
     */
    return new ThreadSafeIterator();
  }

  private class ThreadSafeIterator implements Iterator<T> {
    @Override
    public boolean hasNext() {
      /**
       * Always true.
       */
      return true;
    }

    private int pickNext() {
      // Just grab one atomically.
      int pick = i.incrementAndGet();
      // Reset to zero if it has exceeded - but no spin, let "just someone" manage it.
      int actual = pick % supplier.size();
      if (pick != actual) {
        // So long as someone has a success before we overflow int we're good.
        i.compareAndSet(pick, actual);
      }
      return actual;
    }

    @Override
    public T next() {
      return supplier.get(pickNext());
    }

    @Override
    public void remove() {
      throw new UnsupportedOperationException("Remove not supported.");
    }

  }

}

NB: Я немного скорректировал код для размещения обоих сценариев. Вы можете взять Iterator за поток или поделиться одним Iterator по потокам.

Ответ 2

У вас есть часть информации ( "кто-нибудь принял этот Pair уже?" ), который должен быть общим для всех потоков. Так что в общем случае вы застряли. Однако, если у вас есть представление об этом размере вашего массива и количестве потоков, вы можете использовать ведра, чтобы сделать его менее болезненным.

Предположим, что мы знаем, что будет 1,000,000 элементов массива и 1000 потоков. Назначьте каждому потоку диапазон (поток # 1 получает элементы 0-999 и т.д.). Теперь вместо 1000 потоков, претендующих на один AtomicInteger, вы не можете иметь никаких утверждений вообще!

Это работает, если вы можете быть уверены, что все ваши потоки будут работать примерно в том же темпе. Если вам нужно обработать случай, когда иногда поток # 1 занят другими вещами, а поток # 2 неактивен, вы можете немного изменить свой образец ковша: в каждом ковше есть AtomicInteger. Теперь потоки, как правило, будут бороться только с собой, но если их ведро пустое, они могут перейти к следующему ведру.

Ответ 3

У меня возникли проблемы с пониманием того, что проблема, которую вы пытаетесь решить, есть?

Выполняет ли каждый поток всю коллекцию?

Является ли беспокойство, что ни одна из двух потоков не может работать на одной и той же Паре одновременно? Но каждый поток должен обрабатывать каждую пару в коллекции?

Или вы хотите, чтобы сбор обрабатывался один раз, используя все потоки?

Ответ 4

Есть одна ключевая вещь, которая неясна в вашем примере - что именно означает это значение?

Порядок элементов не имеет значения, поток может принимать один и тот же элемент в другое время.

"разное время" означает что? В течение N миллисекунд друг от друга? Означает ли это, что абсолютно два потока никогда не будут касаться одной и той же пары одновременно? Я буду считать, что.

Если вы хотите уменьшить вероятность того, что потоки будут блокировать друг друга, сражаясь за ту же пару, и есть массив подкреплений пар, попробуйте следующее:

• Разделите свой массив на под-массивы numPairs / threadCount (вам не нужно создавать суб-массивы, просто начинайте с разных смещений, но легче думать о них как о подматрице)
• Назначьте каждый поток другому подматрицу; когда поток исчерпывает свою подматрицу, увеличивайте индекс своего под массива
  • Скажем, у нас есть 6 пар и 2 потока - ваши задания выглядят как Thread-1: [0,1,2] Thread-2: [3,4,5]. Когда начинается Thread-1, он будет рассматривать другой набор пар, чем поток 2, поэтому маловероятно, что они будут бороться за ту же пару.
• Если важно, чтобы два потока на самом деле не касались пары одновременно, заверните весь код, который касается объекта Pair в synchronized(pair) (синхронизируйте экземпляр, а не тип!) - иногда может быть заблокированным, но вы никогда не блокируете все потоки ни на одну вещь, так как с помощью AtomicInteger - потоки могут блокировать друг друга, потому что они действительно пытаются коснуться одного и того же объекта.

Обратите внимание, что это не гарантировано никогда не блокировать - для этого все потоки должны выполняться с одинаковой скоростью, и обработка каждого объекта Pair должна занимать ровно столько же времени, и планировщик потоков ОС должен будет никогда не крать время из одного потока, но не другого. Вы не можете принять любую из этих вещей. То, что это дает вам, - это более высокая вероятность того, что вы добьетесь улучшения concurrency, разделив области на работу и сделав блокировку самой маленькой единичной единицы.

Но это обычный шаблон для получения более concurrency в структуре данных - разделяйте данные между потоками, чтобы они редко касались одной и той же блокировки одновременно.

Ответ 5

Самое простое, что можно увидеть, - создать набор хэшей или карту и дать уникальный хэш для каждого потока. После этого просто попробуйте получить этот хэш-код.

Ответ 6

Это стандартная проблема использования семафора java. Следующий javadoc дает почти аналогичный пример вашей проблемы. http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html

Если вам нужна дополнительная помощь, дайте мне знать?

Ответ 7

Я предпочитаю процесс блокировки и выпуска.

Если поток запрашивает парный объект, объект Pair удаляется из поставщика. Перед тем, как поток запрашивает новую пару, "старая" пара добавляется снова в suplier.

Вы можете нажать спереди и положить в конец.