指尖上的代码--之ACTIVEMQ（4）

FailoverTransport—Transport重连机制

已经找到了开启和完成任务的地方，那么到底在哪里调用开启工作任务的呢？

搜寻FailoverTransport发现，

一个在FailoverTransport构造函数创建TaskRunner时传入的Task匿名对象中反调TaskRunner的wakeup方法。它是在前面讲到的完成重连工作的doReconnect()失败后，再次唤醒重连所用，这是一个一旦唤醒后，不需外力自我校正重连的过程。

另外一个在重连方法reconnect中。而调用reconnect的有，add和start方法。add在前面createTransport(CompositeData compositData)创建FailoverTransport中就有调用，但此时不会唤醒重连，因为此时Transport还未start。那么一般来说第一次reconnect是在start方法中调用的。

    public void reconnect(boolean rebalance) {

        synchronized (reconnectMutex) {

            if (started) {

                if (rebalance) {

                    doRebalance = true;

                }

                LOG.debug("Waking up reconnect task");

                try {

                    reconnectTask.wakeup();

                } catch (InterruptedException e) {

                    Thread.currentThread().interrupt();

                }

            } else {

                LOG.debug("Reconnect was triggered but transport is not started yet. Wait for start to connect the transport.");

            }

        }

}

还有一个是在处理Transport出错的时候。事实也是如此， TransportListener中处理异常的onException方法和发送消息的oneway方法中处理IO异常那里调用了handleTransportFailure方法。

    public final void handleTransportFailure(IOException e) throws InterruptedException {

        if (LOG.isTraceEnabled()) {

            LOG.trace(this + " handleTransportFailure: " + e);

        }

        Transport transport = connectedTransport.getAndSet(null);

        if (transport == null) {

            // sync with possible in progress reconnect

            synchronized (reconnectMutex) {

                transport = connectedTransport.getAndSet(null);

            }

        }

        if (transport != null) {

 

            disposeTransport(transport);

 

            boolean reconnectOk = false;

            synchronized (reconnectMutex) {

                if (started) {

                    LOG.warn("Transport (" + transport.getRemoteAddress() + ") failed to " + connectedTransportURI

                            + " , attempting to automatically reconnect due to: " + e);

                    LOG.debug("Transport failed with the following exception:", e);

                    reconnectOk = true;

                }

                initialized = false;

                failedConnectTransportURI = connectedTransportURI;

                connectedTransportURI = null;

                connected = false;

 

                // notify before any reconnect attempt so ack state can be

                // whacked

                if (transportListener != null) {

                    transportListener.transportInterupted();

                }

 

                if (reconnectOk) {

                    reconnectTask.wakeup();

                }

            }

        }

    }

 

接下来看一下FailoverTransport是如何完成失败后恢复重连的，

final boolean doReconnect() {

    Exception failure = null;

    synchronized (reconnectMutex) {

 

        //第一部分是用来处理配置形式uris的，如果有配置，优先读取配置

        //的uri进行重连，这里没有用到配置。

        String fileURL = getUpdateURIsURL();

        if (fileURL != null) {

            ……

        }

 

        if (disposed || connectionFailure != null) {

            reconnectMutex.notifyAll();

        }

 

        if ((connectedTransport.get() != null && !doRebalance) || disposed || connectionFailure != null) {

            return false;

        } else {

            List<URI> connectList = getConnectList();

            if (connectList.isEmpty()) {

                failure = new IOException("No uris available to connect to.");

            } else {

              //第二部分是做重连负载的，这里也没有用到

                if (doRebalance) {

                   if (connectList.get(0).equals(connectedTransportURI)) {

                            // already connected to first in the list, no need to rebalance

                            doRebalance = false;

                            return false;

                        } else {

                            LOG.debug("Doing rebalance from: " + connectedTransportURI + " to " + connectList);

                            try {

                                Transport transport = this.connectedTransport.getAndSet(null);

                                if (transport != null) {

                                    disposeTransport(transport);

                                }

                            } catch (Exception e) {

                                LOG.debug("Caught an exception stopping existing transport for rebalance", e);

                            }

                        }

                }

                if (!useExponentialBackOff || reconnectDelay == DEFAULT_INITIAL_RECONNECT_DELAY) {

                    reconnectDelay = initialReconnectDelay;

                }

                synchronized (backupMutex) {

                  //第三部分是关于Transport的备份机制的，这里也没有设置备份

                    if (backup && !backups.isEmpty()) {

                        ……

                    }

                }

             

              //第四部分生成新的Transport，进行重连。

                Iterator<URI> iter = connectList.iterator();

                while (iter.hasNext() && connectedTransport.get() == null && !disposed) {

                    URI uri = iter.next();

                    Transport t = null;

                    try {

                        LOG.debug("Attempting connect to: " + uri);

                        SslContext.setCurrentSslContext(brokerSslContext);

                        t = TransportFactory.compositeConnect(uri);

                        t.setTransportListener(myTransportListener);

                        t.start();

 

                        if (started) {

                            restoreTransport(t);

                        }

 

                        LOG.debug("Connection established");

                        reconnectDelay = initialReconnectDelay;

                        connectedTransportURI = uri;

                        connectedTransport.set(t);

                        reconnectMutex.notifyAll();

                        connectFailures = 0;

                        // Make sure on initial startup, that the

                        // transportListener

                        // has been initialized for this instance.

                        synchronized (listenerMutex) {

                            if (transportListener == null) {

                                try {

                                    // if it isn't set after 2secs - it

                                    // probably never will be

                                    listenerMutex.wait(2000);

                                } catch (InterruptedException ex) {

                                }

                            }

                        }

                        ……

                }

            }

        }

             ……

            return false;

        }

    }

    if (!disposed) {

       ……

    }

    return !disposed;

}

第一部分是用来处理配置形式uris的，如果有配置，优先读取配置的uri，并添加到重连uris中本身不做重连，这里没有用到该配置。

第二部分是做负载重连的，根据重连uris如果第一个已经对应的Transport在工作了则无需重连直接返回，否则去掉当前工作的Transport达到负载的目的。这里也没有用到。

第三部分是关于Transport的备份机制的，如果设置了备份机制，且有Transport已经备份则取出该备份Transport返回。这里也没有设置备份。

可以看到前三部分都没有进行实际的重连工作，第四部分才是在上述三部分都不存在的情况下，进行实际的重连工作。

重连工作其实也很简单，根据uri找到对应的Transport，对该Transport设置独有的myTransportListener并启动。然后设置到FailoverTransport的connectedTransport作为当前连接的Transport。

 

通过对FailoverTransport重连机制的分析，基本上可以回答上面提出的第一个问题，为什么合成模式中不是直接add Transport而是uri，并且是数组形式的uris？因为对于重连而言合成Transport是没有意义的，一旦连接异常，作为被合成的Transport对象自身没有重连机制，注定要被废弃。那么只有把uri添加进来，才对后续的重连有益。而数组形式的uris则增强了重连的功能，使其能够尝试不同的uri进行重连。