Синхронизировать в DataOutputStream

Если бы это был я.....

У меня будет LinkedBlockingQueue для каждого поток на стороне клиента. Затем каждый раз, когда у клиентского потока возникает момент бездействия в сокете, он проверяет очередь. Если есть сообщение для отправки из очереди, оно отправляется.

Затем сервер, если это необходимо, может просто добавить элементы в эту очередь, и, когда в соединении появится свободное место, оно будет отправлено.

Добавьте очередь, используйте метод ServerThread, например:

addBroadcastMessage(MyData data) {
    broadcastQueue.add(data);
}

а затем, на стороне сокета, есть цикл с блоком тайм-аута, чтобы он вырывался из сокета, если он простаивает, а затем просто:

while (!broadcastQueue.isEmpty()) {
    MyData data = broadcastQueue.poll();
    .... send the data....
}

и вы сделали.

LinkedBlockingQueue будет управлять блокировкой и синхронизацией для вас.

Ты на правильном пути.

Каждое выражение, изменяющее DataOutputStream, должно быть synchronized в этом DataOutputStream, чтобы к нему не осуществлялся одновременный доступ (и, следовательно, не было каких-либо одновременных изменений):

public void replyToOne(String reply){
    try {
        synchronized(commandOut) {    // writing block
            commandOut.writeUTF(reply);
            commandOut.flush();
        }
    } catch (IOException e) {
        logger.fatal("replyToOne", e);
    }
    logger.info(reply);
}

И:

public static void distribute(String broadcastMessage){
    for (Map.Entry<String, Object[]> entry : AccountInfoList.entrySet()) {
        Object[] tmpObjArray = entry.getValue();
        DataOutputStream temporaryCOut = (DataOutputStream) tmpObjArray[INT_COMMAND_OUT]; //can be grabbed while thread is using it
        try {
            synchronized(temporaryCOut) {  // writing block
                temporaryCOut.writeUTF(broadcastMessage);
                temporaryCOut.flush();
            }
        } catch (IOException e) {
            logger.error("distribute: writeUTF", e);
        }
    logger.info(broadcastMessage);  
    }
}

Просто вставлю свои 2 цента:

Я реализую серверы следующим образом:

Каждый сервер представляет собой поток, выполняющий только одну задачу: прослушивание соединений. Как только он распознает соединение, он создает новый поток для обработки ввода/вывода соединения (я называю этот подкласс ClientHandler).

Сервер также хранит список всех подключенных клиентов.

ClientHandlers отвечают за взаимодействие пользователя с сервером. Отсюда все довольно просто:

Отказ от ответственности: здесь нет блоков try-catches! добавьте их самостоятельно. Конечно, вы можете использовать исполнители потоков, чтобы ограничить количество одновременных подключений.

Метод сервера run():

@Override
public void run(){
 isRunning = true;
 while(isRunning){
  ClientHandler ch = new ClientHandler(serversocket.accept());
  clients.add(ch);
  ch.start();
 }
}

Ctor ClientHandler:

public ClientHandler(Socket client){
 out = new ObjectOutputStream(client.getOutputStream());
 in = new ObjectInputStream(client.getInputStream());
}

Метод run() ClientHandler:

@Override
public void run(){
 isConnected = true;
 while(isConnected){
  handle(in.readObject());
 }
}

и handle() метод:

private void handle(Object o){
 //Your implementation
}

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

Есть 2 простых способа сделать это:

1. Оберните каждый вызов для вывода в синхронизированный (этот) блок
2. Используйте геттер для вывода (как и вы) с ключевым словом synchronized.