将会话/事务的超时设置为DefaultMessageListenerContainer [英] Set timeOut for session/transaction to DefaultMessageListenerContainer

问题描述

我想使用Spring的DefaultMessageListenerConatiner(通过重写doReceiveAndExecute)在一段时间内(从接收到第一条消息后300毫秒)从Queue接收消息块，如

I want to receive chunk of messages from Queue within some timeLimit(Ex : 300 millisec after receiving the 1st message) using DefaultMessageListenerConatiner Of Spring (By overriding doReceiveAndExecute) as mentioned in the link.

当队列中有太多消息时，我可以将批量大小为20的消息分组，而当队列中的消息很少时，我可以接收不到20条消息.

I can group the messages of my batch size i.e 20 when the queue is having too many messages and I can receive less than 20 messages when there are very less messages in Queue.

问题:

我看到即使队列已满，将消息发送到侦听器也要花费太多时间(有时1秒，有时2秒甚至更多).

I see it takes too much time(sometimes 1 sec and sometime 2 secs and more) for sending the messages to Listener even when the queue is full.

当我尝试使用DefaultMessageListenerConatiner并发接收单个消息时，我看到消息的接收延迟为几毫秒(例如1毫秒或最多30到60毫秒)

When I try with DefaultMessageListenerConatiner as such to receive single messages concurrently, I see the messages are received in a delay of few milliseconds(like 1 millisec or max 30 to 60 millisec)

我没有指定transactionTimeout或receiveTimeout，也没有链接任何transactionManager.

I didn't specify transactionTimeout or receiveTimeout and I didn't link any transactionManager as well.

施普林格能否帮助我找到可以指定超时的位置或如何减少时间延迟?

Can Springers please help me to find where the timeOut can be specified or How can I redeuce the time delay?

BatchMessageListenerContainer:

package com.mypackage;

import javax.jms.Session;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;

import java.util.ArrayList;
import java.util.List;

import org.springframework.jms.listener.DefaultMessageListenerContainer;

import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.support.JmsUtils;
import org.springframework.transaction.TransactionStatus;

/**
 * Listener Container that allows batch consumption of messages. Works only with transacted sessions
 */
public class BatchMessageListenerContainer extends DefaultMessageListenerContainer {
  public static final int DEFAULT_BATCH_SIZE = 20;

  private int batchSize = DEFAULT_BATCH_SIZE;

  public BatchMessageListenerContainer() {
    super();
    setSessionTransacted(true);
  }

  /**
   * @return The batch size on this container
   */
  public int getBatchSize() {
    return batchSize;
  }

  /**
   * @param batchSize The batchSize of this container
   */
  public void setBatchSize(int batchSize) {
    this.batchSize = batchSize;
  }

  /**
   * The doReceiveAndExecute() method has to be overriden to support multiple-message receives.
   */
  @Override
  protected boolean doReceiveAndExecute(Object invoker, Session session, MessageConsumer consumer,
    TransactionStatus status) throws JMSException {

    Connection conToClose = null;
    MessageConsumer consumerToClose = null;
    Session sessionToClose = null;

    try {
      Session sessionToUse = session;
      MessageConsumer consumerToUse = consumer;

      if (sessionToUse == null) {
        Connection conToUse = null;
        if (sharedConnectionEnabled()) {
          conToUse = getSharedConnection();
        }
        else {
          conToUse = createConnection();
          conToClose = conToUse;
          conToUse.start();
        }
        sessionToUse = createSession(conToUse);
        sessionToClose = sessionToUse;
      }

      if (consumerToUse == null) {
        consumerToUse = createListenerConsumer(sessionToUse);
        consumerToClose = consumerToUse;
      }

      List<Message> messages = new ArrayList<Message>();

      int count = 0;
      Message message = null;
      // Attempt to receive messages with the consumer
      do {
        message = receiveMessage(consumerToUse);
        if (message != null) {
          messages.add(message);
        }
      }
      // Exit loop if no message was received in the time out specified, or
      // if the max batch size was met
      while ((message != null) && (++count < batchSize));

      if (messages.size() > 0) {
        // Only if messages were collected, notify the listener to consume the same.
        try {
          doExecuteListener(sessionToUse, messages);
          sessionToUse.commit();
        }
        catch (Throwable ex) {
          handleListenerException(ex);
          if (ex instanceof JMSException) {
            throw (JMSException) ex;
          }
        }
        return true;
      }

      // No message was received for the period of the timeout, return false.
      noMessageReceived(invoker, sessionToUse);
      return false;
    }
    finally {
      JmsUtils.closeMessageConsumer(consumerToClose);
      JmsUtils.closeSession(sessionToClose);
      ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), true);
    }
  }

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

    @SuppressWarnings("unchecked")
    SessionAwareBatchMessageListener<Message> lsnr = (SessionAwareBatchMessageListener<Message>) getMessageListener();

    try {
      lsnr.onMessages(session, messages);
    }
    catch (JMSException ex) {
      rollbackOnExceptionIfNecessary(session, ex);
      throw ex;
    }
    catch (RuntimeException ex) {
      rollbackOnExceptionIfNecessary(session, ex);
      throw ex;
    }
    catch (Error err) {
      rollbackOnExceptionIfNecessary(session, err);
      throw err;
    }
  }

  @Override
  protected void checkMessageListener(Object messageListener) {
    if (!(messageListener instanceof SessionAwareBatchMessageListener<?>)) {
      throw new IllegalArgumentException("Message listener needs to be of type ["
        + SessionAwareBatchMessageListener.class.getName() + "]");
    }
  }

  @Override
  protected void validateConfiguration() {
    if (batchSize <= 0) {
      throw new IllegalArgumentException("Property batchSize must be a value greater than 0");
    }
  }

  public void setSessionTransacted(boolean transacted) {
    if (!transacted) {
      throw new IllegalArgumentException("Batch Listener requires a transacted Session");
    }
    super.setSessionTransacted(transacted);
  }
}

SessionAwareBatchMessageListener:

package com.mypackage;

import java.util.List;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

public interface SessionAwareBatchMessageListener<M extends Message> {
  /**
   * Perform a batch action with the provided list of {@code messages}.
   * 
   * @param session JMS {@code Session} that received the messages
   * @param messages List of messages
   * @throws JMSException JMSException thrown if there is an error performing the operation.
   */
  public void onMessages(Session session, List<M> messages) throws JMSException;
}

applicationContext.xml中的Bean:

<bean id="myMessageListener" class="org.mypackage.MyMessageListener">

<bean id="jmsContainer" class="com.mypackage.BatchMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory"/>
    <property name="destinationName" ref="queue"/>
    <property name="messageListener" ref="myMessageListener"/>
    <property name ="concurrentConsumers" value ="10"/>
    <property name ="maxConcurrentConsumers" value ="50"/>        
</bean>

MyMessageListner:

MyMessageListner :

package org.mypackage;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;

import org.mypackage.service.MyService;

public class MyMessageListener implements SessionAwareBatchMessageListener<TextMessage> {

    @Autowired
    private MyService myService;

    @Override
    public void onMessage(Session session, List<TextMessage> messages) {
        try {
           for(TextMessage tm :messages)  {
                TextMessage textMessage = (TextMessage) message;
               // parse the message and add to list
            }
            //process list of Objects to DB
        } catch (JMSException e1) {
             e1.printStackTrace();
        }
    }
}

推荐答案

我认为在向消费者发送消息之前所花费的时间是您的while循环引起的，因为您每次都在等待列表填满，但是由于该线程是在doReceiveAndExecute方法内部创建的，因此只能由当前线程填充！

i think that the time spent before sending the message to the consumer was caused by your while loop, because you're awaiting each time the list to be full but this one is only filled by the current thread since it is created inside the doReceiveAndExecute method!

  // Exit loop if no message was received in the time out specified, or
  // if the max batch size was met
  while ((message != null) && (++count < batchSize));

也许这可以做得很好:

...
     List<Message> messages = Collections.synchronizedList(new ArrayList<Message>());

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

        Connection conToClose = null;
        MessageConsumer consumerToClose = null;
        Session sessionToClose = null;

        try {
          Session sessionToUse = session;
          MessageConsumer consumerToUse = consumer;

          if (sessionToUse == null) {
            Connection conToUse = null;
            if (sharedConnectionEnabled()) {
              conToUse = getSharedConnection();
            }
            else {
              conToUse = createConnection();
              conToClose = conToUse;
              conToUse.start();
            }
            sessionToUse = createSession(conToUse);
            sessionToClose = sessionToUse;
          }

          if (consumerToUse == null) {
            consumerToUse = createListenerConsumer(sessionToUse);
            consumerToClose = consumerToUse;
          }


          Message message = null;
          // Attempt to receive messages with the consumer
          do {
            message = receiveMessage(consumerToUse);
            if (message != null) {
              messages.add(message);
            }
          }

          if (messages.size() >= batchSize)) {
            synchronized (messages) {
              // Only if messages were collected, notify the listener to consume the same.
              try {
                doExecuteListener(sessionToUse, messages);
                sessionToUse.commit();
                // clear the list!!
                messages.clear();
              }
              catch (Throwable ex) {
                handleListenerException(ex);
                if (ex instanceof JMSException) {
                  throw (JMSException) ex;
                }
              }
            }
            return true;
          }

          // No message was received for the period of the timeout, return false.
          noMessageReceived(invoker, sessionToUse);
          return false;
        }
        finally {
          JmsUtils.closeMessageConsumer(consumerToClose);
          JmsUtils.closeSession(sessionToClose);
          ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), true);
        }
      }

这篇关于将会话/事务的超时设置为DefaultMessageListenerContainer的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！