<h3>0 .概述</h3>    <p>最近，公司的项目需要重构代码，因为个人最近接触了Rx这个高大上的东东，</p>    <p>而，项目原来是满满的回调地狱...项目是基于Socket的长连接，所以呢，</p>    <p>从最底层开始，把普通的Socket撸成Rx!</p>    <h3>1 .设计模式</h3>    <p>首先，这个RxSocket是唯一的，也就是，全局唯一咯，</p>    <p>嗯，设计成单例。</p>    <pre>  <code class="language-java">public static RxSocket getInstance() {          RxSocket rxSocket = defaultInstance;          if (defaultInstance == null) {              synchronized (RxSocket.class) {                  rxSocket = defaultInstance;                  if (defaultInstance == null) {                      rxSocket = new RxSocket();                      defaultInstance = rxSocket;                  }              }          }          return rxSocket;      }</code></pre>    <p>双重加锁型的单例。</p>    <h3>2 .对外的接口/方法</h3>    <p>Socket，第一个想到就是连接，读写，而，我们外界想知道的，就只是是否</p>    <p>写，连接是否成功，和读到啥数据。所以定义：</p>    <pre>  <code class="language-java">public Observable<Boolean> connectRx(String ip, int port);  public Observable<Boolean> disConnect();  public Observable<byte[]> read();  public Observable<Boolean> write(ByteBuffer buffer);</code></pre>    <p>还有一点，应该要有一个方法，让外界知道，这个Socket的状态，也就是监听方法：</p>    <p>public Observable<SocketStatus> socketStatusListener ();</p>    <h3>3 .具体代码实现</h3>    <pre>  <code class="language-java">package chestnut.RxSocket;    import java.net.InetSocketAddress;  import java.nio.ByteBuffer;  import java.nio.channels.SelectionKey;  import java.nio.channels.Selector;  import java.nio.channels.SocketChannel;  import java.util.Iterator;  import java.util.concurrent.TimeUnit;    import chestnut.utils.LogUtils;  import rx.Observable;  import rx.Subscriber;  import rx.schedulers.Schedulers;  import rx.subjects.PublishSubject;  import rx.subjects.SerializedSubject;  import rx.subjects.Subject;    /**   * Created by Chestnut on 2016/12/18.   */    public class RxSocket {        //本类的常量      private static final String TAG = "RxSocket";      private static final boolean OpenLog = true;      private static final long WRITE_TIME_OUT = 3000;      private static final long CONNECT_TIME_OUT = 3000;        //单例      private Subject<Object,byte[]> readSubject;      private Subject<Object,SocketStatus> connectStatus;      private static volatile RxSocket defaultInstance;      private RxSocket() {          readSubject = new SerializedSubject(PublishSubject.create());          connectStatus = new SerializedSubject(PublishSubject.create());      }      public static RxSocket getInstance() {          RxSocket rxSocket = defaultInstance;          if (defaultInstance == null) {              synchronized (RxSocket.class) {                  rxSocket = defaultInstance;                  if (defaultInstance == null) {                      rxSocket = new RxSocket();                      defaultInstance = rxSocket;                  }              }          }          return rxSocket;      }        //变量      private SocketStatus socketStatus = SocketStatus.DIS_CONNECT;      private Selector selector = null;      private SocketChannel socketChannel = null;      private SelectionKey selectionKey = null;      private ReadThread readThread = null;      private boolean isReadThreadAlive = true;      private SocketReconnectCallback socketReconnectCallback = null;        //方法        /**       * 监听Socket的状态       * @return       */      public Observable<SocketStatus> socketStatusListener () {          return connectStatus;      }        /**       * 建立Socket连接，只是尝试建立一次       * @param ip    IP or 域名       * @param port  端口       * @return  Rx true or false       */      public Observable<Boolean> connectRx(String ip, int port) {          return Observable                  .create(new Observable.OnSubscribe<Boolean>() {                      @Override                      public void call(Subscriber<? super Boolean> subscriber) {                            //正在连接                          if (socketStatus == SocketStatus.CONNECTING) {                              subscriber.onNext(false);                              subscriber.onCompleted();                              return;                          }                            //未连接 | 已经连接，关闭Socket                          socketStatus = SocketStatus.DIS_CONNECT;                          isReadThreadAlive = false;                          readThread = null;                          if (selector!=null)                              try {                                  selector.close();                              } catch (Exception e) {                                  LogUtils.i(OpenLog,TAG,"selector.close");                              }                          if (selectionKey!=null)                              try {                                  selectionKey.cancel();                              } catch (Exception e) {                                  LogUtils.i(OpenLog,TAG,"selectionKey.cancel");                              }                          if (socketChannel!=null)                              try {                                  socketChannel.close();                              } catch (Exception e) {                                  LogUtils.i(OpenLog,TAG,"socketChannel.close");                              }                            //重启Socket                          isReadThreadAlive = true;                          readThread = new ReadThread(ip,port);                          readThread.start();                          socketReconnectCallback = new SocketReconnectCallback() {                              @Override                              public void onSuccess() {                                  subscriber.onNext(true);                                  subscriber.onCompleted();                              }                                @Override                              public void onFail(String msg) {                                  LogUtils.i(OpenLog,TAG,"connectRx:"+msg);                                  subscriber.onNext(false);                                  subscriber.onCompleted();                              }                          };                      }                  })                  .subscribeOn(Schedulers.newThread())                  .timeout(CONNECT_TIME_OUT, TimeUnit.MILLISECONDS, Observable.just(false));      }        /**       * 断开当前的Socket       * @return Rx true or false       */      public Observable<Boolean> disConnect() {          return Observable.create(new Observable.OnSubscribe<Boolean>() {              @Override              public void call(Subscriber<? super Boolean> subscriber) {                  try {                      if (socketStatus == SocketStatus.DIS_CONNECT) {                          subscriber.onNext(true);                          subscriber.onCompleted();                      }                      else {                          socketStatus = SocketStatus.DIS_CONNECT;                          isReadThreadAlive = false;                          readThread = null;                          if (selector!=null)                              try {                                  selector.close();                              } catch (Exception e) {                                  LogUtils.i(OpenLog,TAG,"selector.close");                              }                          if (selectionKey!=null)                              try {                                  selectionKey.cancel();                              } catch (Exception e) {                                  LogUtils.i(OpenLog,TAG,"selectionKey.cancel");                              }                          if (socketChannel!=null)                              try {                                  socketChannel.close();                              } catch (Exception e) {                                  LogUtils.i(OpenLog,TAG,"socketChannel.close");                              }                          subscriber.onNext(true);                          subscriber.onCompleted();                      }                  } catch (Exception e) {                      subscriber.onNext(false);                      subscriber.onCompleted();                  }              }          });      }        /**       * 读取Socket的消息       * @return  Rx error 或者 有数据       */      public Observable<byte[]> read() {          if (socketStatus != SocketStatus.CONNECTED)              return Observable.create(new Observable.OnSubscribe<byte[]>() {                  @Override                  public void call(Subscriber<? super byte[]> subscriber) {                      subscriber.onError(new Throwable("Socket Dis Connect"));                  }              });          else              return readSubject;      }        /**       * 向Socket写消息       * @param buffer    数据包       * @return  Rx true or false       */      public Observable<Boolean> write(ByteBuffer buffer) {          return Observable                  .create(new Observable.OnSubscribe<Boolean>() {                      @Override                      public void call(Subscriber<? super Boolean> subscriber) {                          if (socketStatus != SocketStatus.CONNECTED) {                              subscriber.onNext(false);                              subscriber.onCompleted();                          }                          else {                              if (socketChannel!=null && socketChannel.isConnected()) {                                  try {                                      int result = socketChannel.write(buffer);                                      if (result==0) {                                          LogUtils.i(OpenLog,TAG,"write."+"服务器断开链接");                                      }                                      else if (result<0) {                                          LogUtils.e(OpenLog, TAG, "write." + "发送出错");                                      }                                      else {                                          subscriber.onNext(true);                                          subscriber.onCompleted();                                      }                                  } catch (Exception e) {                                      LogUtils.i(OpenLog,TAG,"write."+e.getMessage());                                      subscriber.onNext(false);                                      subscriber.onCompleted();                                  }                              }                              else {                                  LogUtils.i(OpenLog,TAG,"write."+"close");                                  subscriber.onNext(false);                                  subscriber.onCompleted();                              }                          }                      }                  })                  .subscribeOn(Schedulers.newThread())                  .timeout(WRITE_TIME_OUT, TimeUnit.MILLISECONDS, Observable.just(false));      }        //类      private class ReadThread extends Thread {          private String ip;          private int port;          ReadThread(String ip, int port) {              this.ip = ip;              this.port = port;          }          @Override          public void run() {              LogUtils.i(OpenLog,TAG,"ReadThread:"+"start");              while (isReadThreadAlive) {                  //连接                  if (socketStatus == SocketStatus.DIS_CONNECT) {                      try {                          if (selectionKey != null) selectionKey.cancel();                          socketChannel = SocketChannel.open();                          socketChannel.configureBlocking(false);                          selector = Selector.open();                          socketChannel.connect(new InetSocketAddress(ip, port));                          selectionKey = socketChannel.register(selector, SelectionKey.OP_CONNECT);                          socketStatus = SocketStatus.CONNECTING;                          connectStatus.onNext(SocketStatus.CONNECTING);                      } catch (Exception e) {                          isReadThreadAlive = false;                          socketStatus = SocketStatus.DIS_CONNECT;                          connectStatus.onNext(SocketStatus.DIS_CONNECT);                          LogUtils.e(OpenLog, TAG, "ReadThread:init:" + e.getMessage());                          if (socketReconnectCallback!=null)                              socketReconnectCallback.onFail("SocketConnectFail1");                      }                  }                  //读取                  else if (socketStatus == SocketStatus.CONNECTING || socketStatus  == SocketStatus.CONNECTED) {                      try {                          selector.select();                          Iterator<SelectionKey> it = selector.selectedKeys().iterator();                          while (it.hasNext()) {                              SelectionKey key = it.next();                              if (key.isConnectable()) {                                  if (socketChannel.isConnectionPending()) {                                      try {                                          socketChannel.finishConnect();                                          socketStatus = SocketStatus.CONNECTED;                                          connectStatus.onNext(SocketStatus.CONNECTED);                                          socketChannel.configureBlocking(false);                                          socketChannel.register(selector, SelectionKey.OP_READ);                                          if (socketReconnectCallback!=null)                                              socketReconnectCallback.onSuccess();                                      } catch (Exception e) {                                          isReadThreadAlive = false;                                          socketStatus = SocketStatus.DIS_CONNECT;                                          connectStatus.onNext(SocketStatus.DIS_CONNECT);                                          LogUtils.e(OpenLog, TAG, "ReadThread:finish:" + e.getMessage());                                          if (socketReconnectCallback!=null)                                              socketReconnectCallback.onFail("SocketConnectFail2");                                      }                                  }                              } else if (key.isReadable()) {                                  ByteBuffer buf = ByteBuffer.allocate(10000);                                  int length = socketChannel.read(buf);                                  if (length <= 0) {                                      LogUtils.e(OpenLog, TAG, "服务器主动断开链接！");                                  } else {                                      byte[] bytes = new byte[length];                                      for (int i = 0; i < length; i++) {                                          bytes[i] = buf.get(i);                                      }                                      readSubject.onNext(bytes);                                  }                              }                          }                          it.remove();                      } catch (Exception e) {                          isReadThreadAlive = false;                          socketStatus = SocketStatus.DIS_CONNECT;                          connectStatus.onNext(SocketStatus.DIS_CONNECT);                          LogUtils.e(OpenLog, TAG, "ReadThread:read:" + e.getMessage());                          if (socketReconnectCallback!=null)                              socketReconnectCallback.onFail("SocketConnectFail3");                      }                  }              }          }      }        //枚举 && 接口      public enum SocketStatus {          DIS_CONNECT,          CONNECTING,          CONNECTED,      }        private interface SocketReconnectCallback {          void onSuccess();          void onFail(String msg);      }    }</code></pre>    <p>额，好像有点长，这个Socket是NIO包的Socket，里面只是开启了一条线程。</p>    <h3>4 .注意</h3>    <ul>     <li> <p>之所以放出一个监听方法，我想的是，Socket连接上后，有可能会被断开，</p> <p>这样，就需要做一个重连的策略，当然，这个策略看项目的要求，</p> <p>因而，我把其对外开放了。你可以监听这个方法，去做Socket的重连策略。</p> </li>     <li> <p>RxSokcet的读方法，需要注意，要在适当的时候去解除订阅。</p> <p>还有，Socket状态的监听也是。</p> </li>     <li> <p>最后，有哪些不合理的地方，各位大老要好好教导一下小弟~</p> </li>    </ul>    <p> </p>    <p>来自：http://www.jianshu.com/p/27e4f714cfa3</p>    <p> </p>