Как сделать поток до тех пор, пока переменная не достигнет определенного значения (многопоточная Java)

Question

У меня есть серверная программа, которая принимает клиентские соединения. Эти клиентские соединения могут принадлежать многим потокам. Например, два или более клиентов могут принадлежать одному и тому же потоку. Из этих потоков я должен пройти одно сообщение, но мне нужно подождать, пока все потоки не будут установлены. Для этого я поддерживаю следующую структуру данных.

ConcurrentHashMap<Integer, AtomicLong> conhasmap = new ConcurrentHashMap<Integer, AtomicLong>(); 

Целое - это идентификатор потока и длинный номер клиента. Чтобы сделать один поток для данного потока ждать, пока AtomicLong не достигнет определенного значения, я использовал следующий цикл. Фактически первый пакет потока помещает идентификатор потока и количество ожидающих соединений. С каждым соединением я уменьшаю соединения для ожидания.

while(conhasmap.get(conectionID) != new AtomicLong(0)){ 
     // Do nothing 
} 

Однако этот цикл блокирует другие потоки. В соответствии с этим answer он выполняет изменчивое считывание. Как я могу изменить код для ожидания правильного потока для данного потока до тех пор, пока он не достигнет определенного значения?

Answer 1

Ваше выражение:

conhasmap.get(conectionID) != new AtomicLong(0) 

всегда будет верным, потому что вы сравниваете ссылки на объекты, которые никогда не будут равны, а не значений. Чем лучше выражение будет:

conhasmap.get(conectionID).longValue() != 0L) 

, но цикл, как это без ожидания/уведомит логики внутри цикла не является хорошей практикой, поскольку она использует время процессора постоянно. Вместо этого каждый поток должен вызывать .wait() в экземпляре AtomicLong, а когда он уменьшался или увеличивался, вы должны вызвать .notifyAll() в экземпляре AtomicLong, чтобы разбудить каждый ожидающий поток, чтобы проверить выражение. Класс AtomicLong уже может вызывать метод notifyAll() всякий раз, когда он изменяется, но я точно не знаю.

AtomicLong al = conhasmap.get(conectionID); 
synchronized(al) { 
    while(al.longValue() != 0L) { 
     al.wait(100); //wait up to 100 millis to be notified 
    } 
} 

В коде, что приращение/декремент, это будет выглядеть следующим образом:

AtomicLong al = conhasmap.get(conectionID); 
synchronized(al) { 
    if(al.decrementAndGet() == 0L) { 
     al.notifyAll(); 
    } 
} 

лично я бы не использовать AtomicLong для этого счетчика, потому что вы не выгоды от блокировки меньше поведения AtomicLong. Просто используйте java.lang.Long вместо этого, потому что вам все равно нужно синхронизировать объект-счетчик для логики wait()/notify().

Answer 2

Если вы используете Java 8, CompletableFuture может быть в хорошей форме. Вот полный, надуманный пример, который ожидает, что 5 клиентов смогут подключиться и отправить сообщение на сервер (имитируется с использованием BlockingQueue с предложением/опросом).

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

В этом примере у вас нет сложных ожидающих/уведомляющих установок потока или циклов ожидания ожидания.

package so.thread.state; 

import java.util.concurrent.*; 
import java.util.concurrent.atomic.AtomicBoolean; 
import java.util.concurrent.atomic.AtomicLong; 

public class Main { 

    public static String CONNETED_MSG = "CONNETED"; 
    public static Long EXPECTED_CONN_COUNT = 5L; 

    public static ExecutorService executor = Executors.newCachedThreadPool(); 
    public static BlockingQueue<String> queue = new LinkedBlockingQueue<>(); 

    public static AtomicBoolean done = new AtomicBoolean(false); 

    public static void main(String[] args) throws Exception { 

    // add a "server" thread 
    executor.submit(() -> server()); 

    // add 5 "client" threads 
    for (int i = 0; i < EXPECTED_CONN_COUNT; i++) { 
     executor.submit(() -> client()); 
    } 

    // clean shut down 
    Thread.sleep(TimeUnit.SECONDS.toMillis(1)); 
    done.set(true); 
    Thread.sleep(TimeUnit.SECONDS.toMillis(1)); 
    executor.shutdown(); 
    executor.awaitTermination(1, TimeUnit.SECONDS); 

    } 

    public static void server() { 

    print("Server started up"); 
    // track # of client connections established 
    AtomicLong connectionCount = new AtomicLong(0L); 

    // at startup, create my "hook" 
    CompletableFuture<Long> hook = new CompletableFuture<>(); 
    hook.thenAcceptAsync(Main::allClientsConnected, executor); 

    // consume messages 
    while (!done.get()) { 
     try { 
     String msg = queue.poll(5, TimeUnit.MILLISECONDS); 
     if (null != msg) { 
      print("Server received client message"); 
      if (CONNETED_MSG.equals(msg)) { 
      long count = connectionCount.incrementAndGet(); 

      if (count >= EXPECTED_CONN_COUNT) { 
       hook.complete(count); 
      } 
      } 
     } 

     } catch (Exception e) { 
     e.printStackTrace(); 
     } 
    } 

    print("Server shut down"); 

    } 

    public static void client() { 
    queue.offer(CONNETED_MSG); 
    print("Client sent message"); 
    } 

    public static void allClientsConnected(Long count) { 
    print("All clients connected, count: " + count); 
    } 


    public static void print(String msg) { 
    System.out.println(String.format("[%s] %s", Thread.currentThread().getName(), msg)); 
    } 
} 

Вы получаете выход как этот

[pool-1-thread-1] Server started up 
[pool-1-thread-5] Client sent message 
[pool-1-thread-3] Client sent message 
[pool-1-thread-2] Client sent message 
[pool-1-thread-6] Client sent message 
[pool-1-thread-4] Client sent message 
[pool-1-thread-1] Server received client message 
[pool-1-thread-1] Server received client message 
[pool-1-thread-1] Server received client message 
[pool-1-thread-1] Server received client message 
[pool-1-thread-1] Server received client message 
[pool-1-thread-4] All clients connected, count: 5 
[pool-1-thread-1] Server shut down