Совместимость с Java LinkedList класс

Question

Good Day,

У меня проблемы параллелизма с LinkedList на Java. У меня есть тип объекта, называемый «Connection», который имеет в качестве переменной-члена LinkedList «MessageHandlers», называемый «прослушиватели». Затем у меня есть два разных потока, один из которых изменяется и один итерации по одному и тому же LinkedList.

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

Exception in thread "Thread-1" java.util.ConcurrentModificationException 

. Есть ли у кого-нибудь другие предложения попробовать? Вот некоторые snipbits моего кода ...

public synchronized Object ReadObject() throws java.io.IOException 
{ 
    Object obj = null; 

    try 
    { 
     obj = input.readObject(); 

     synchronized(listeners) 
     { 
      Iterator<MessageHandler> i = listeners.iterator(); 

      while(i.hasNext()) 
      { 
       i.next().MessageReceived(obj, this); 
      } 
     } 
    } 
    catch (IOException e) 
    { 
     e.printStackTrace(); 
     throw e; 
    } 
    catch (ClassNotFoundException e) 
    { 
     // TODO Auto-generated catch block 
     e.printStackTrace(); 
    } 

    return obj; 
} 

Вышеприведенный код находится внутри моего объекта соединения. Он вызывается из функции, которая имеет данные чтения ObjectInputStream сокета из сокета. «Input» - это экземпляр ObjectInputStream.

public void addNewLoggedInUser(User user) throws Exception 
{ 
    for(User u:loggedInUsers) 
    { 
     if(u == user) 
     { 
      throw new Exception("That user is already logged in"); 
     } 
    } 

    //Add the new users 
    loggedInUsers.add(user); 

    synchronized(user.getConnection().getListeners()) 
    { 
     user.getConnection().getListeners().add(this); 
    } 

    this.SendGameStatusUpdateToAllLoggedinPlayers(); 
} 

Я затем вызвать метод user.getConnection(). GetListeners(). Добавить (это) и, таким образом, я получаю исключение.

public Connection() 
{ 
    //Initialize the variables to NULL 
    socket    = null; 
    output    = null; 
    input    = null; 
    receiveThread  = null; 
    runReceiveThread = false; 
    listeners   = Collections.synchronizedList(new LinkedList<MessageHandler>()); 

    //Handle the ID counter. Now we have a unique ID for every connection that comes in 
    connectionID = counterID; 
    counterID = counterID + 1; 
} 

Это конструктор класса соединений. Обратите внимание, что он Collections.synchronizedList

Любые идеи? Большое спасибо за Вашу помощь!

Answer 1

Блоки synchronized выглядят так, как будто они должны работать. Я ожидал бы, что есть активность в методах, вызванных из блока ReadObjectsynchronized, которые изменяют список. Сделайте ли какой-либо из ваших звонков или цепочек MessageHandler вызов на номер addNewLoggedInUser (или любой другой способ, который может обновить список слушателей)?

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

Answer 2

java.util.ConcurrentModificationException на самом деле не проблема с резьбой. Это связано с изменением списка, заблокированного его итератором. Думаю, вы звоните addNewLoggedInUser() от MessageReceived(). Это вызовет исключение параллельной модификации, поскольку вызывающая функция уже имеет блокировку итератора в связанном списке.

Answer 3

Пройдите через BlockingQueue javadoc. Он упоминает простой сценарий, как хорошо, что соответствует вашим требованиям, т.е.

class Producer implements Runnable { 
    private final BlockingQueue queue; 
    Producer(BlockingQueue q) { queue = q; } 
    public void run() { 
    try { 
     while (true) { queue.put(produce()); } 
    } catch (InterruptedException ex) { ... handle ...} 
    } 
    Object produce() { ... } 
} 

class Consumer implements Runnable { 
    private final BlockingQueue queue; 
    Consumer(BlockingQueue q) { queue = q; } 
    public void run() { 
    try { 
     while (true) { consume(queue.take()); } 
    } catch (InterruptedException ex) { ... handle ...} 
    } 
    void consume(Object x) { ... } 
} 

class Setup { 
    void main() { 
    BlockingQueue q = new SomeQueueImplementation(); 
    Producer p = new Producer(q); 
    Consumer c1 = new Consumer(q); 
    Consumer c2 = new Consumer(q); 
    new Thread(p).start(); 
    new Thread(c1).start(); 
    new Thread(c2).start(); 
    } 
}