ActiveMQ启动阻塞问题排查

灰太狼 2022-05-06 07:25 961阅读 0赞

## ActiveMQ启动阻塞问题排查 ##

首先，看一看有问题的代码。基于activemq-client-5.14.5.jar封装的启动方法：

public void start() throws NotifyException {
            try {
                if (this.connection == null) {
                    if (this.clientId == null || this.clientId.length() < 1) {
                        throw new NotifyException("the clientId is null");
                    }
    
                    this.connection = this.connectionFactory.createConnection();
                    this.connection.setClientID(this.clientId);
                }
    
                LOG.debug("-  start connection AMQ");
                Thread thread = new Thread(new Runnable() {
                    public void run() {
                        try {
                            ActiveMqConnection.this.connection.start();
                        } catch (JMSException var2) {
                            ActiveMqConnection.LOG.debug("-  start connection AMQ error[{}]", var2.getMessage());
                        }
    
                    }
                });
                thread.setName("notify-consumer-start:" + this.clientId);
                thread.start();
            } catch (JMSException var2) {
                LOG.debug("-  start connection AMQ error[{}]", var2.getMessage());
            }
    
        }

首先创建connection、设置clientId，然后启动。看似没毛病。如果activemq未开启，在createConnection这一步就抛出异常了。但是，这次遇到的问题却是createConnection成功了，在setClientId这一步阻塞了。  
ActiveMQConnection中setClientId

public void setClientID(String newClientID) throws JMSException {
            this.checkClosedOrFailed();
            if (this.clientIDSet) {
                throw new IllegalStateException("The clientID has already been set");
            } else if (this.isConnectionInfoSentToBroker) {
                throw new IllegalStateException("Setting clientID on a used Connection is not allowed");
            } else {
                this.info.setClientId(newClientID);
                this.userSpecifiedClientID = true;
                this.ensureConnectionInfoSent();
            }
        }

阻塞在了ensureConnectionInfoSent这个方法中，该方法会发送一个包到activemq的服务器，检查该连接。

protected void ensureConnectionInfoSent() throws JMSException {
            Object var1 = this.ensureConnectionInfoSentMutex;
            synchronized(this.ensureConnectionInfoSentMutex) {
                if (!this.isConnectionInfoSentToBroker && !this.closed.get()) {
                    if (this.info.getClientId() == null || this.info.getClientId().trim().length() == 0) {
                        this.info.setClientId(this.clientIdGenerator.generateId());
                    }
    
                    this.syncSendPacket(this.info.copy(), this.getConnectResponseTimeout());
                    this.isConnectionInfoSentToBroker = true;
                    ConsumerId consumerId = new ConsumerId(new SessionId(this.info.getConnectionId(), -1L), this.consumerIdGenerator.getNextSequenceId());
                    if (this.watchTopicAdvisories) {
                        this.advisoryConsumer = new AdvisoryConsumer(this, consumerId);
                    }
    
                }
            }
        }

一步一步断点进入之后，在FailoverTransport中发现一个叫oneway的方法，该方法中有一个死循序

while(true) {
                            if (!this.disposed) {
                                try {
                                    Transport transport = (Transport)this.connectedTransport.get();
                                    long start = System.currentTimeMillis();
    
                                    boolean timedout;
                                    for(timedout = false; transport == null && !this.disposed && this.connectionFailure == null && !Thread.currentThread().isInterrupted() && this.willReconnect(); transport = (Transport)this.connectedTransport.get()) {
                                        LOG.trace("Waiting for transport to reconnect..: {}", command);
                                        long end = System.currentTimeMillis();
                                        if (command.isMessage() && this.timeout > 0L && end - start > this.timeout) {
                                            timedout = true;
                                            LOG.info("Failover timed out after {} ms", end - start);
                                            break;
                                        }
    
                                        try {
                                            this.reconnectMutex.wait(100L);
                                        } catch (InterruptedException var16) {
                                            Thread.currentThread().interrupt();
                                            LOG.debug("Interupted:", var16);
                                        }
                                    }
    
                                    if (transport != null) {
                                        Tracked tracked = null;
    
                                        try {
                                            tracked = this.stateTracker.track(command);
                                        } catch (IOException var15) {
                                            LOG.debug("Cannot track the command {} {}", command, var15);
                                        }
    
                                        Map var11 = this.requestMap;
                                        synchronized(this.requestMap) {
                                            if (tracked != null && tracked.isWaitingForResponse()) {
                                                this.requestMap.put(command.getCommandId(), tracked);
                                            } else if (tracked == null && command.isResponseRequired()) {
                                                this.requestMap.put(command.getCommandId(), command);
                                            }
                                        }
    
                                        try {
                                            transport.oneway(command);
                                            this.stateTracker.trackBack(command);
                                            if (command.isShutdownInfo()) {
                                                this.shuttingDown = true;
                                            }
                                        } catch (IOException var17) {
                                            if (tracked == null && this.canReconnect()) {
                                                if (command.isResponseRequired()) {
                                                    this.requestMap.remove(command.getCommandId());
                                                }
    
                                                throw var17;
                                            }
    
                                            LOG.debug("Send oneway attempt: {} failed for command: {}", i, command);
                                            this.handleTransportFailure(var17);
                                        }
                                        break;
                                    }
    
                                    if (this.disposed) {
                                        error = new IOException("Transport disposed.");
                                    } else if (this.connectionFailure != null) {
                                        error = this.connectionFailure;
                                    } else if (timedout) {
                                        error = new IOException("Failover timeout of " + this.timeout + " ms reached.");
                                    } else if (!this.willReconnect()) {
                                        error = new IOException("Reconnect attempts of " + this.maxReconnectAttempts + " exceeded");
                                    } else {
                                        error = new IOException("Unexpected failure.");
                                    }
                                } catch (IOException var19) {
                                    LOG.debug("Send oneway attempt: {} failed for command: {}", i, command);
                                    this.handleTransportFailure(var19);
                                    ++i;
                                    continue;
                                }
                            }
                            break label202;
                        }

该方法会判断向activemq服务器的请求是否超时，未超时就一直死循环。就是卡在了这个地方，因为在自己封装的启动方法并未设置超时时间，这里timeout的值一直是0。  
最后，改为设置超时时间，并将setClientId方法也放到线程中执行。

public void start() throws NotifyException {
            try {
                if (this.connection == null) {
                    if (this.clientId == null || this.clientId.length() < 1) {
                        throw new NotifyException("the clientId is null");
                    }
    
                    this.connection = this.connectionFactory.createConnection();
                    setTimeOut(30000);
                }
    
                LOG.debug("-  start connection AMQ");
                Thread thread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            connection.setClientID(clientId);
                            ActiveNotifyConnection.this.connection.start();
                        } catch (JMSException var2) {
                            LogUtil.debug(ActiveNotifyConnection.class, "start connection AMQ error");
                        }
    
                    }
                });
                thread.setName("notify-consumer-start:" + this.clientId);
                thread.start();
            } catch (JMSException var2) {
                LOG.debug("-  start connection AMQ error[{}]", var2.getMessage());
            }
    
        }

其实，就原来的代码，重启完activemq之后也是正常的，至于为什么会出现这种情况依然无法解释。