原创

Java观察者设计模式导致死循环

笔者使用观察者模式成功在线上环境跑出了死循环,差点造成了车毁人亡的惨痛教训,写本篇文章方便以后查看。

1.观察者模式演示

定义对象间一对多的依赖关系,当一个对象的状态发生改变时,所有依赖于它的对象都得到通知自动更新。

ObserverTest.java

package cn.gameboys;

import java.util.Enumeration;
import java.util.Vector;

public class ObserverTest {

 // 观察者模式
 // 定义对象间一对多的依赖关系,当一个对象的状态发生改变时,
 // 所有依赖于它的对象都得到通知自动更新。
 public static void main(String[] args) {
  Subject sub = new MySubject();
  sub.add(new Observer1());
  sub.add(new Observer2());
  sub.operation();
 }

}

interface Observer {
 public void update();
}

class Observer1 implements Observer {

 @Override
 public void update() {
  System.out.println("observer1 has received!");
 }
}

class Observer2 implements Observer {

 @Override
 public void update() {
  System.out.println("observer2 has received!");
 }

}

interface Subject {

 /* 增加观察者 */
 public void add(Observer observer);

 /* 删除观察者 */
 public void del(Observer observer);

 /* 通知所有的观察者 */
 public void notifyObservers();

 /* 自身的操作 */
 public void operation();
}

abstract class AbstractSubject implements Subject {

 private Vector<Observer> vector = new Vector<Observer>();

 @Override
 public void add(Observer observer) {
  vector.add(observer);
 }

 @Override
 public void del(Observer observer) {
  vector.remove(observer);
 }

 @Override
 public void notifyObservers() {
  Enumeration<Observer> enumo = vector.elements();
  while (enumo.hasMoreElements()) {
   enumo.nextElement().update();
  }
 }
}

class MySubject extends AbstractSubject {
 @Override
 public void operation() {
  System.out.println("update self!");
  notifyObservers();
 }
}

2.观察者模式并发出现死循环

笔者设计某个需求的时候,正好使用到了这种设计模式,使用期间,由于未考虑到自引用的问题,导致了 BUG,照成死循环,代码如下

package cn.gameboys;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;

public class BadObserverManeger {

 public static void main(String[] args) {

  List<BadObservable> list = new ArrayList<BadObservable>();
  for (int i = 0; i < 10; i++) {
   BadObservable ob = new BadObserver();
   list.add(ob);
   BadObserverManeger.getIns().addPublisher(1, ob);
  }
  try {
   BadObserverManeger.getIns().notifyObserver(1, list);

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

  }

 }

 private static BadObserverManeger _ins = new BadObserverManeger();

 private BadObserverManeger() {

 }

 public static BadObserverManeger getIns() {
  return _ins;
 }

 private Map<Integer, Vector<BadObservable>> typeBadObservableMap = new ConcurrentHashMap<Integer, Vector<BadObservable>>();

 public void addPublisher(int type, BadObservable notify) {
  Vector<BadObservable> vec = typeBadObservableMap.get(type);
  if (vec == null) {
   vec = new Vector<BadObservable>();
   typeBadObservableMap.put(type, vec);
  }
  synchronized (this) {
   vec.add(notify);
  }
 }

 public void remove(int type, BadObservable notify) {
  Vector<BadObservable> vec = typeBadObservableMap.get(type);
  if (vec != null) {
   synchronized (this) {
    vec.remove(notify);
   }
  }
 }

 public void notifyObserver(int type, Object param) {
  Vector<BadObservable> set = typeBadObservableMap.get(type);
  if (set != null) {
   synchronized (this) {


    //for循环和while循环会进入死循环
    for (Iterator<BadObservable> it = set.iterator(); it.hasNext();) {
     try {
      try {
       Thread.currentThread().sleep(1000);
      } catch (InterruptedException e) {
       e.printStackTrace();
      }
      it.next().notify(type, param);
     } catch (Exception e) {
      e.printStackTrace();
     }
    }


/*    Iterator<BadObservable> it = set.iterator();
    while (it.hasNext()) {
     try {
      try {
       Thread.currentThread().sleep(1000);
      } catch (InterruptedException e) {
       e.printStackTrace();
      }
      it.next().notify(type, param);
     } catch (Exception e) {
      e.printStackTrace();
     }
    }*/



   //foreach不会进入死循环?为啥呢?
/*    for (BadObservable badObservable : set) {
     try {
      try {
       Thread.currentThread().sleep(1000);
      } catch (InterruptedException e) {
       e.printStackTrace();
      }
      badObservable.notify(type, param);
     } catch (Exception e) {
      e.printStackTrace();
     }
    }*/


   }
  }
 }

}

interface BadObservable {
 void notify(int type, Object param);
}

class BadObserver implements BadObservable {

 @Override
 public void notify(int type, Object param) {
  System.out.println("notify");

  List<BadObservable> list = (List<BadObservable>) param;

  BadObserverManeger.getIns().remove(1, list.get(2));

 }
}

class TestBadUseThread extends Thread {

 List<BadObservable> list;

 public TestBadUseThread(List<BadObservable> list) {
  this.list = list;
 }

 @Override
 public void run() {

  try {
   BadObserverManeger.getIns().notifyObserver(1, list);

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

  for (int i = 0; i < 100; i++) {
   try {
    Thread.currentThread().sleep(1000);
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }

 }

}


执行结果:

执行结果
执行结果

注意上面代码中的 notifyObserver 方法,笔者首次使用的是迭代器模式,更新正式服出 BUG 之后导致疯狂写日志,半小时写了 150G,差点填表回家~(开玩笑) 上面三种方式,while,迭代器,foreach 三种里面,foreach 神奇般的没有死循环!!!,为啥呢????TMD 难道 foreach 和 iterator 不是一样?笔者进入了深入研究。。。

3.foreach 和 Iterator 结果为啥会不一样?

为了探究 foreach 和迭代器的区别,小编做了下面两个方法

foreach

 public static void main(String[] args) {
  Set<Integer> set = new HashSet<Integer>();
  set.add(1);
  set.add(2);
  for (Integer i : set) {
   try {
    System.out.println(i);
   } catch (Exception e) {
    e.printStackTrace();
   }
  }
 }

iterator

 public static void main(String[] args) {
  Set<Integer> set = new HashSet<Integer>();
  set.add(1);
  set.add(2);
  for (Iterator<Integer> it = set.iterator(); it.hasNext();) {
   try {
    System.out.println(it.next());
   } catch (Exception e) {
    e.printStackTrace();
   }
  }
 }

笔者通过javap -verbose ForEachTest看了一下 jvm 字节码文件,结果如下

对比结果
对比结果

从上面图中,可以解除的疑惑是 foreach 和 iteartor 都是一样的,底层都是调用的 iteartor,只是执行顺序不一样,另外,重点来了!!!!

捕获异常的行数不一样,foreach 捕获的是 50-57,注意看,Iterator.next:()不在捕获异常范围。。。。再看 iterator,捕获异常是从 40 到 52 行,里面包含了 Iterator.next:()方法。。

真相大白,就是因为我们在 iterator 里面 for 循环内加了 tryCatch,导致逻辑报错了但是循环未跳出去,而 foreach 里面 for 循环内部报错会抛错到 for 外层,跳出了 for 循环。

4.观察者模式并发无 BUG 优化版

其实上面示例有两个 bug,1 是死循环,另外一个是并发操作 hashMap,导致报错,那我们怎么处理 notify 的时候不反向操作 map 呢?

笔者期间想了好几个方案,其中有一个就是讲 remove 再抛异步,延时处理,也能解决这个问题,不过不好的是需要修改观察者,极其不合理,

笔者在咨询多方之后,在某位大佬处得到一个牛逼的思想,不删,保存状态,最后一期删。

代码就是为每个 BadObservable 增加一个状态,数据结构如下:

private Map<Integer, Map<BadObservable,Boolean>> typeBadObservableMap = new ConcurrentHashMap<Integer, ConcurrentHashMap<BadObservable,Boolean>>();

实现代码就不具体贴了,很简单。至此,笔者的两个疑问都解决了。。

总结

  • 观察者模式使用场景为:定义对象间一对多的依赖关系,当一个对象的状态发生改变时,所有依赖于它的对象都得到通知自动更新。
  • foreach 和 iteartor 的差别仅仅是抛异常的地方不一样,foreach 抛异常是在 for 外部。
  • 解决观察者模式 push 的时候反向调用删除观察者。可采用记录状态,集中删除。
正文到此结束