Spring----监听器容器

Posted 2021-02-05 AoTuDeMan

消息监听器容器是一个用于查看JMS目标等待消息到达的特殊的bean，一旦消息到达它就可以获取到消息，并通过调用onMessage方法将消息传递一个MessageListener实现。Spring中消息监听器容器的类型如下：

　　❤ SimpleMessageListenerContainer：最简单的消息监听器容器，只能处理固定数量的JMS会话，且不支持事务。

　　❤ DefaultMessageListenerContainer：这个消息监听器容器建立在SimpleMessageListenerContainer容器之上，添加了对事物的支持。

　　❤ serversession.ServerSessionMessage.ListenerContainer：这是功能最强大的消息监听器，与DefaultMessageListenerContainer相同，它支持事务，但是它还允许动态地管理JMS会话。

下面以DefaultMessageListenerContainer为例进行分析，看看消息监听器容器的实现。使用消息监听器容器时一定要将自定义的消息监听器置于到容器中，这样才可以在收到消息时，容器把消息转向监听器处理。下面看一下它的类图：

同样我们看到了此类实现了InitializingBean接口，按照以往的风格我们还是首先查看接口方法afterPropertiesSet中的逻辑，其方法的实现在其父类AbstractJmsListeningContainer中。

public void afterPropertiesSet() {
        //验证connectionFactory
        super.afterPropertiesSet();
        //验证配置文件
        validateConfiguration();
        //初始化
        initialize();
    }

监听器容器的初始化只包含了三句代码，其中前两句只用于属性的验证，而真正用于初始化的操作委托在initialize中执行。

public void initialize() throws JmsException {
        try {
            synchronized (this.lifecycleMonitor) {
                this.active = true;
                this.lifecycleMonitor.notifyAll();
            }
            doInitialize();
        }
        catch (JMSException ex) {
            synchronized (this.sharedConnectionMonitor) {
                ConnectionFactoryUtils.releaseConnection(this.sharedConnection, getConnectionFactory(), this.autoStartup);
                this.sharedConnection = null;
            }
            throw convertJmsAccessException(ex);
        }
    }

protected void doInitialize() throws JMSException {
        synchronized (this.lifecycleMonitor) {
            for (int i = 0; i < this.concurrentConsumers; i++) {
                scheduleNewInvoker();
            }
        }
    }

这里用到了concurrentConsumers属性，对于此属性的说明如下：

　　消息监听器允许创建多个Session和MessageConsumer来接收消息。具体的个数由concurrentConsumers属性指定。需要注意的是，应该只是在Destination为Queue的时候才使用多个MessageConsumer（Queue中的一个消息只能被一个Consumer接收），虽然使用多个MessageConsumer会提高消息的处理性能，但是消息处理的顺序不能得到保证。消息被接收的顺序仍然是消息发送时的顺序，但是由于消息可能被并发处理，因此，消息的处理顺序可能和消息发送顺序不同，此外，不应该在Destination为Topic的时候使用多个MessageConsumer，因为多个MessageConsumer会接收到同样的消息。

对于具体的实现逻辑我们只能继续查看源码：

private void scheduleNewInvoker() {
        AsyncMessageListenerInvoker invoker = new AsyncMessageListenerInvoker();
        if (rescheduleTaskIfNecessary(invoker)) {
            // This should always be true, since we\'re only calling this when active.
            this.scheduledInvokers.add(invoker);
        }
    }

protected final boolean rescheduleTaskIfNecessary(Object task) {
        if (this.running) {
            try {
                doRescheduleTask(task);
            }
            catch (RuntimeException ex) {
                logRejectedTask(task, ex);
                this.pausedTasks.add(task);
            }
            return true;
        }
        else if (this.active) {
            this.pausedTasks.add(task);
            return true;
        }
        else {
            return false;
        }
    }

 分析源码得知，根据concurrentConsumers数量建立了对应数量的线程，而每个线程都作为一个独立的接收者在循环接收消息。

反向追踪rescheduleTaskIfNecessary传入的参数invoker，发现这个参数是AsyncMessageListenerInvoker类型的，于是我们把焦点转向这个类的实现，由于它是作为一个Runnable角色去执行，所以我们从这个类的分析从run方法开始。

public void run() {
            //并发控制
            synchronized (lifecycleMonitor) {
                activeInvokerCount++;
                lifecycleMonitor.notifyAll();
            }
            boolean messageReceived = false;
            try {
                //根据每个任务设置的最大处理消息数量而做不同处理，小于0默认为是无限制，一致接收消息
                if (maxMessagesPerTask < 0) {
                    messageReceived = executeOngoingLoop();
                }
                else {
                    int messageCount = 0;
                    //消息数量控制，一旦超出数量则停止循环
                    while (isRunning() && messageCount < maxMessagesPerTask) {
                        messageReceived = (invokeListener() || messageReceived);
                        messageCount++;
                    }
                }
            }
            catch (Throwable ex) {
                //清理操作，包括关闭Session等
                clearResources();
                if (!this.lastMessageSucceeded) {
                    // We failed more than once in a row or on startup -
                    // wait before first recovery attempt.
                    waitBeforeRecoveryAttempt();
                }
                this.lastMessageSucceeded = false;
                boolean alreadyRecovered = false;
                synchronized (recoveryMonitor) {
                    if (this.lastRecoveryMarker == currentRecoveryMarker) {
                        handleListenerSetupFailure(ex, false);
                        recoverAfterListenerSetupFailure();
                        currentRecoveryMarker = new Object();
                    }
                    else {
                        alreadyRecovered = true;
                    }
                }
                if (alreadyRecovered) {
                    handleListenerSetupFailure(ex, true);
                }
            }
            finally {
                synchronized (lifecycleMonitor) {
                    decreaseActiveInvokerCount();
                    lifecycleMonitor.notifyAll();
                }
                if (!messageReceived) {
                    this.idleTaskExecutionCount++;
                }
                else {
                    this.idleTaskExecutionCount = 0;
                }
                synchronized (lifecycleMonitor) {
                    if (!shouldRescheduleInvoker(this.idleTaskExecutionCount) || !rescheduleTaskIfNecessary(this)) {
                        // We\'re shutting down completely.
                        scheduledInvokers.remove(this);
                        if (logger.isDebugEnabled()) {
                            logger.debug("Lowered scheduled invoker count: " + scheduledInvokers.size());
                        }
                        lifecycleMonitor.notifyAll();
                        clearResources();
                    }
                    else if (isRunning()) {
                        int nonPausedConsumers = getScheduledConsumerCount() - getPausedTaskCount();
                        if (nonPausedConsumers < 1) {
                            logger.error("All scheduled consumers have been paused, probably due to tasks having been rejected. " +
                                    "Check your thread pool configuration! Manual recovery necessary through a start() call.");
                        }
                        else if (nonPausedConsumers < getConcurrentConsumers()) {
                            logger.warn("Number of scheduled consumers has dropped below concurrentConsumers limit, probably " +
                                    "due to tasks having been rejected. Check your thread pool configuration! Automatic recovery " +
                                    "to be triggered by remaining consumers.");
                        }
                    }
                }
            }
        }

run

以上函数主要根据变量maxMessagePerTask的值来分为不同的情况处理，当然，函数中还使用了大量的代码异常处理机制的数据维护。

其实核心的处理就是调用invokeListener来接收消息并激活消息监听器，但是之所以两种情况分开处理，正是考虑到了在无限制的循环接收消息的情况下，用户可以设置标志位running来控制消息的接收的暂停与恢复，并维护当前消息监听器的数量。

private boolean executeOngoingLoop() throws JMSException {
            boolean messageReceived = false;
            boolean active = true;
            while (active) {
                synchronized (lifecycleMonitor) {
                    boolean interrupted = false;
                    boolean wasWaiting = false;
                    //如果当前任务已经处于激活状态但是却给了暂时中止的命令
                    while ((active = isActive()) && !isRunning()) {
                        if (interrupted) {
                            throw new IllegalStateException("Thread was interrupted while waiting for " +
                                    "a restart of the listener container, but container is still stopped");
                        }
                        if (!wasWaiting) {
                            //如果并非处于等待状态则说明第一次执行，需要将激活的任务数量减少
                            decreaseActiveInvokerCount();
                        }
                        //开始进入等待状态，等待任务的恢复命令
                        wasWaiting = true;
                        try {
                            //通过wait等待，也就是等待notify或者notifyAll
                            lifecycleMonitor.wait();
                        }
                        catch (InterruptedException ex) {
                            // Re-interrupt current thread, to allow other threads to react.
                            Thread.currentThread().interrupt();
                            interrupted = true;
                        }
                    }
                    if (wasWaiting) {
                        activeInvokerCount++;
                    }
                    if (scheduledInvokers.size() > maxConcurrentConsumers) {
                        active = false;
                    }
                }
                //正常处理流程
                if (active) {
                    messageReceived = (invokeListener() || messageReceived);
                }
            }
            return messageReceived;
        }

如果按照正常的流程其实是不会进入while循环的，而是直接进入函数invokeListener来接收消息并激活监听器，但是，我们不可能让循环一直持续下去，我们要考虑到暂停线程或者恢复线程的情况，这时，isRunning函数就派上用场了。

isRunning用来检测标志位this.running状态而判断是否需要进入while循环。由于要维护当前线程的激活数量，所以引入了wasWaiting变量，用来判断线程是否处理等待状态。如果线程首次进入等待状态，则需要减少线程激活数量计数器。

当然，还有个地方需要提一下，就是线程等待不是一味的采用while循环来控制，因为如果单纯的采用while循环会浪费CPU的始终周期，给资源造成巨大的浪费。这里，Spring采用的是全局控制变量LifecycleMonitor的wait方法来暂停线程，所以，如果终止线程需要再次恢复的话，除了更改this.running标志位外，还需要调用LifecycleMonitor.notify或者LifecycleMonitor.notifyAll来使得线程恢复。

接下来就是消息接收的处理了。

private boolean invokeListener() throws JMSException {
            this.currentReceiveThread = Thread.currentThread();
            try {
                //初始化资源包括首次创建的时候创建Session与Consumer
                initResourcesIfNecessary();
                boolean messageReceived = receiveAndExecute(this, this.session, this.consumer);
                //改变标志位，信息成功处理
                this.lastMessageSucceeded = true;
                return messageReceived;
            }
            finally {
                this.currentReceiveThread = null;
            }
        }

protected boolean receiveAndExecute(
            Object invoker, @Nullable Session session, @Nullable MessageConsumer consumer)
            throws JMSException {

        if (this.transactionManager != null) {
            // Execute receive within transaction.
            TransactionStatus status = this.transactionManager.getTransaction(this.transactionDefinition);
            boolean messageReceived;
            try {
                messageReceived = doReceiveAndExecute(invoker, session, consumer, status);
            }
            catch (JMSException | RuntimeException | Error ex) {
                rollbackOnException(this.transactionManager, status, ex);
                throw ex;
            }
            this.transactionManager.commit(status);
            return messageReceived;
        }

        else {
            // Execute receive outside of transaction.
            return doReceiveAndExecute(invoker, session, consumer, null);
        }
    }

由于DefaultMessageListenerContainer消息监听器容器建立在SimpleMessageListenerContainer容器之上，添加了对事务的支持，那么此时，事务特性已经开始了。如果用户设置了this.transcationManager，也就是配置了事务，那么消息的接收会被控制在事务之内，一旦出现任何异常都会回滚，而回滚操作也会交于事务管理器统一处理。

doReceiveAndExecute包含了整个消息的接收处理过程，由于参杂着事务，所以并没有复用模板中的方法。

protected boolean doReceiveAndExecute(Object invoker, @Nullable Session session,
            @Nullable MessageConsumer consumer, @Nullable TransactionStatus status) throws JMSException {

        Connection conToClose = null;
        Session sessionToClose = null;
        MessageConsumer consumerToClose = null;
        try {
            Session sessionToUse = session;
            boolean transactional = false;
            if (sessionToUse == null) {
                sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(
                        obtainConnectionFactory(), this.transactionalResourceFactory, true);
                transactional = (sessionToUse != null);
            }
            if (sessionToUse == null) {
                Connection conToUse;
                if (sharedConnectionEnabled()) {
                    conToUse = getSharedConnection();
                }
                else {
                    conToUse = createConnection();
                    conToClose = conToUse;
                    conToUse.start();
                }
                sessionToUse = createSession(conToUse);
                sessionToClose = sessionToUse;
            }
            MessageConsumer consumerToUse = consumer;
            if (consumerToUse == null) {
                consumerToUse = createListenerConsumer(sessionToUse);
                consumerToClose = consumerToUse;
            }
            //接收消息
            Message message = receiveMessage(consumerToUse);
            if (message != null) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Received message of type [" + message.getClass() + "] from consumer [" +
                            consumerToUse + "] of " + (transactional ? "transactional " : "") + "session [" +
                            sessionToUse + "]");
                }
                //模板方法，当消息接收且未处理前给子类机会做相应处理
                messageReceived(invoker, sessionToUse);
                boolean exposeResource = (!transactional && isExposeListenerSession() &&
                        !TransactionSynchronizationManager.hasResource(obtainConnectionFactory()));
                if (exposeResource) {
                    TransactionSynchronizationManager.bindResource(
                            obtainConnectionFactory(), new LocallyExposedJmsResourceHolder(sessionToUse));
                }
                try {
                    //激活监听器
                    doExecuteListener(sessionToUse, message);
                }
                catch (Throwable ex) {
                    if (status != null) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("Rolling back transaction because of listener exception thrown: " + ex);
                        }
                        status.setRollbackOnly();
                    }
                    handleListenerException(ex);
                    // Rethrow JMSException to indicate an infrastructure problem
                    // that may have to trigger recovery...
                    if (ex instanceof JMSException) {
                        throw (JMSException) ex;
                    }
                }
                finally {
                    if (exposeResource) {
                        TransactionSynchronizationManager.unbindResource(obtainConnectionFactory());
                    }
                }
                // Indicate that a message has been received.
                return true;
            }
            else {
                if (logger.isTraceEnabled()) {
                    logger.trace("Consumer [" + consumerToUse + "] of " + (transactional ? "transactional " : "") +
                            "session [" + sessionToUse + "] did not receive a message");
                }
                //接收到空消息的处理
                noMessageReceived(invoker, sessionToUse);
                // Nevertheless call commit, in order to reset the transaction timeout (if any).
                if (shouldCommitAfterNoMessageReceived(sessionToUse)) {
                    commitIfNecessary(sessionToUse, null);
                }
                // Indicate that no message has been received.
                return false;
            }
        }
        finally {
            JmsUtils.closeMessageConsumer(consumerToClose);
            JmsUtils.closeSession(sessionToClose);
            ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), true);
        }
    }

doReceiveAndExecute

 上述函数代码看似复杂，但是真正的逻辑不多，大多是固定的套路，而我们最关心就是监听器的激活。

protected void doExecuteListener(Session session, Message message) throws JMSException {
        if (!isAcceptMessagesWhileStopping() && !isRunning()) {
            if (logger.isWarnEnabled()) {
                logger.warn("Rejecting received message because of the listener container " +
                        "having been stopped in the meantime: " + message);
            }
            rollbackIfNecessary(session);
            throw new MessageRejectedWhileStoppingException();
        }

        try {
            invokeListener(session, message);
        }
        catch (JMSException | RuntimeException | Error ex) {
            rollbackOnExceptionIfNecessary(session, ex);
            throw ex;
        }
        commitIfNecessary(session, message);
    }

protected void invokeListener(Session session, Message message) throws JMSException {
        Object listener = getMessageListener();

        if (listener instanceof SessionAwareMessageListener) {
            doInvokeListener((SessionAwareMessageListener) listener, session, message);
        }
        else if (listener instanceof MessageListener) {
            doInvokeListener((MessageListener) listener, message);
        }
        else if (listener != null) {
            throw new IllegalArgumentException(
                    "Only MessageListener and SessionAwareMessageListener supported: " + listener);
        }
        else {
            throw new IllegalStateException("No message listener specified - see property \'messageListener\'");
        }
    }

通过层层调用，最终提取监听器并使用invokeListener激活了监听器，也就是激活了用户自定义的监听器逻辑。这里还有一句代码commitIfNecessary(session, message)，完成的功能是session.commit()。完成消息服务的事务提交，涉及两个事务，我们常说的DefaultMessageListenerContainer增加了对事务的支持，是通用的事务，也就是说我们在消息接收过程中如果产生其他操作，比如向数据库中插入数据，一旦出现异常时就需要全部回滚，也包括回滚插入数据库中的数据。但是除了我们常说的事务之外，对于消息本身还有一个事务，当接收一个消息的时候，必须使用事务提交的方式，这是在告诉消息服务器本地已经正常接收消息，消息服务器接收到本地的事务提交后便可以将此消息删除，否则，当前消息会被其他接收者重新接收。

参考：《Spring源码深度解析》 郝佳 编著；