Spring----JmsTemplate

尽管消息接收可以使用消息监听的方式替代模板方法，但是发送的时候是无法代替的，在Spring中必须要使用JmsTemplate提供的方法来进行发送操作，可见JmsTemplate类的重要性，那么我们对于整合消息服务的分析就从JmsTemplate开始。

JmsTemplate

在上一篇文章中，我们看到了Spring采用了与JDBC等一贯的套路，为我们提供了JmsTemplate来封装常用操作。下面看一下它的类图：

提取我们感兴趣的接口InitializingBean，接口方法的实现是在JmsAccessor类中，如下：

public void afterPropertiesSet() {
        if (getConnectionFactory() == null) {
            throw new IllegalArgumentException("Property 'connectionFactory' is required");
        }
    }

发现函数中只是一个验证功能，并没有逻辑实现。所以丢掉这里，转向实例代码分析。首先以发送为例，在Spring中发送消息可以通过JmsTemplate中提供的方法来实现。

public void send(final Destination destination, final MessageCreator messageCreator) throws JmsException {
        execute(session -> {
            doSend(session, destination, messageCreator);
            return null;
        }, false);
    }

这个函数的风格不得不让我们回想起JdbcTemplate的类实现风格，极为相似，都是提取一个公共的方法作为最底层、最通用的功能实现，然后又通过回调函数的不同来区分个性化的功能。

我们首先查看通用代码的抽取实现。

1.通用代码抽取

public <T> T execute(SessionCallback<T> action, boolean startConnection) throws JmsException {
        Assert.notNull(action, "Callback object must not be null");
        Connection conToClose = null;
        Session sessionToClose = null;
        try {
            Session sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(
                    obtainConnectionFactory(), this.transactionalResourceFactory, startConnection);
            if (sessionToUse == null) {
                //创建connection
                conToClose = createConnection();
                //创建session
                sessionToClose = createSession(conToClose);
                //是否开启向服务器推送连接信息，只有接收消息时需要，发送时不需要
                if (startConnection) {
                    conToClose.start();
                }
                sessionToUse = sessionToClose;
            }
            if (logger.isDebugEnabled()) {
                logger.debug("Executing callback on JMS Session: " + sessionToUse);
            }
            //调用回调函数
            return action.doInJms(sessionToUse);
        }
        catch (JMSException ex) {
            throw convertJmsAccessException(ex);
        }
        finally {
            //关闭session
            JmsUtils.closeSession(sessionToClose);
            //释放连接
            ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), startConnection);
        }
    }

上述代码封装了connection以及session的创建操作。在准备工作结束后，调用回调函数将程序引入用户自定义实现的个性化处理。

2.发送消息的实现

有了基类的辅助实现，使得Spring更加专注于个性的处理，也就是说Spring使用execute方法中封装了冗余代码，而将个性化的代码实现放在了回调函数doInJms中。在发送消息的功能中回调函数通过局部类实现。

execute(session -> {
            doSend(session, destination, messageCreator);
            return null;
        }, false);

此时的发送逻辑已经完全转向了doSend方法，这样使整个功能实现变得更加清晰了。

protected void doSend(Session session, Destination destination, MessageCreator messageCreator)
            throws JMSException {

        Assert.notNull(messageCreator, "MessageCreator must not be null");
        MessageProducer producer = createProducer(session, destination);
        try {
            Message message = messageCreator.createMessage(session);
            if (logger.isDebugEnabled()) {
                logger.debug("Sending created message: " + message);
            }
            doSend(producer, message);
            // Check commit - avoid commit call within a JTA transaction.
            if (session.getTransacted() && isSessionLocallyTransacted(session)) {
                // Transacted session created by this template -> commit.
                JmsUtils.commitIfNecessary(session);
            }
        }
        finally {
            JmsUtils.closeMessageProducer(producer);
        }
    }

在发送消息还是遵循着消息发送的规则，比如根据Destination创建MessageProducer、创建Message，并使用MessageProducer实例来发送消息。

3.接收消息

我们通常使用JmsTemplate.receive(destination)来接收简单的消息，那么这个功能Spring是如何封装的呢？

public Message receive(Destination destination) throws JmsException {
        return receiveSelected(destination, null);
    }

public Message receiveSelected(final Destination destination, @Nullable final String messageSelector) throws JmsException {
        return execute(session -> doReceive(session, destination, messageSelector), true);
    }

protected Message doReceive(Session session, Destination destination, @Nullable String messageSelector)
            throws JMSException {

        return doReceive(session, createConsumer(session, destination, messageSelector));
    }

protected Message doReceive(Session session, MessageConsumer consumer) throws JMSException {
        try {
            // Use transaction timeout (if available).
            long timeout = getReceiveTimeout();
            ConnectionFactory connectionFactory = getConnectionFactory();
            JmsResourceHolder resourceHolder = null;
            if (connectionFactory != null) {
                resourceHolder = (JmsResourceHolder) TransactionSynchronizationManager.getResource(connectionFactory);
            }
            if (resourceHolder != null && resourceHolder.hasTimeout()) {
                timeout = Math.min(timeout, resourceHolder.getTimeToLiveInMillis());
            }
            Message message = receiveFromConsumer(consumer, timeout);
            if (session.getTransacted()) {
                // Commit necessary - but avoid commit call within a JTA transaction.
                if (isSessionLocallyTransacted(session)) {
                    // Transacted session created by this template -> commit.
                    JmsUtils.commitIfNecessary(session);
                }
            }
            else if (isClientAcknowledge(session)) {
                // Manually acknowledge message, if any.
                if (message != null) {
                    message.acknowledge();
                }
            }
            return message;
        }
        finally {
            JmsUtils.closeMessageConsumer(consumer);
        }
    }

实现的套路还是和发送差不多，同样还是使用execute函数来封装多余的操作，而最终的目的还是通过consumer.receive()来接收消息，其中的过程就是对于MessageConsumer的创建以及一些辅助操作。

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