Duplicated messages on ActiveMQ

Question

I use ActiveMQ as JMS broker and consumer, jmsTemplate to send the messages, 1 non-durable Topic for the moment. During the peak time I have ~100 messages/second.

It doesn't matter how many messages are in the queue, but I frequently get duplicated messages. The temporary solution that I came up is to set up index on table - for the moment all messages are only saved in database.

My first question - why messages are duplicated, if I specified non-durable Topic and the response is not required?

Sender:

@Component
public class QueueSender 
{
    private Logger log = Logger.getLogger(getClass());
@Autowired
    protected JmsTemplate jmsTemplate;


    public JmsTemplate getJmsTemplate() {
        return jmsTemplate;
    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    @Autowired
    public QueueSender( final JmsTemplate jmsTemplate )
    {
        this.jmsTemplate = jmsTemplate;
        this.jmsTemplate.setDeliveryPersistent(false);
        System.out.println("isSessionTransacted "+jmsTemplate.isSessionTransacted()+
                " getDeliveryMode "+jmsTemplate.getDeliveryMode()+
                " getReceiveTimeout "+jmsTemplate.getReceiveTimeout()+
                " getSessionAcknowledgeMode "+jmsTemplate.getSessionAcknowledgeMode());
    }


    public void sendPrice(Integer tickerId, Integer field, Double price, Long timestamp)
    {
        jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        jmsTemplate.setMessageIdEnabled(true);
        Map <String, Object>map = new HashMap<String, Object>();
        map.put("tickerId", tickerId);
        map.put("field", field);
        map.put("price", price);
        map.put("timestamp", timestamp);
        jmsTemplate.convertAndSend("Quotez", map);
    }

    public void sendVolume(Integer tickerId, Integer field, Integer size, Long timestamp)
    {
        jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

        Map <String, Object>map = new HashMap<String, Object>();
        map.put("tickerId", tickerId);
        map.put("field", field);
        map.put("size", size);
        map.put("timestamp", timestamp);
        jmsTemplate.convertAndSend("Quotez", map);

    }

}

Listener:

public void onMessage(Message message) 
{
     if (message instanceof MapMessage) 
     {           
         try
         {
             MapMessage mapMessage = (MapMessage) message;
                 if(null !=  mapMessage.getString("price"))
                 {
 priceService.insert(mapMessage.getInt("tickerId"),mapMessage.getDouble("price"),
mapMessage.getInt("field"),mapMessage.getLong("timestamp"));
                 }                     else{
volumeService.insert(mapMessage.getInt("tickerId"),mapMessage.getInt("size"),
mapMessage.getInt("field"),mapMessage.getLong("timestamp"));
             }
         }
         catch (final JMSException e)
         {
             exceptionListener.onException(e);
         }
     }
}

Spring:

<amq:broker useJmx="true" persistent="false">
<amq:transportConnectors>
  <amq:transportConnector uri="tcp://localhost:0"/>
</amq:transportConnectors> </amq:broker>
<amq:topic id="topicDest"  physicalName="Quotez"/>
  <amq:connectionFactory id="jmsFactory" brokerURL="vm://localhost?jms.watchTopicAdvisories=false"/>  
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<constructor-arg ref="jmsFactory" />
<property name="exceptionListener" ref="jmsExceptionListener" />
<property name="sessionCacheSize" value="100" />
</bean>


<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
   <constructor-arg ref="connectionFactory"/>
    <property name="pubSubDomain" value="true"/>
 <property name="defaultDestinationName" value="Quotez"/>    
</bean>
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="topicDest"/>
        <property name="messageListener" ref="jdbcListener" />
    </bean>

The second question is concerning jmsContainer configuration. What is the difference between the code above and the code below? The code above gives me Topic as subscriber and the code below gives me Queue.

<jms:listener-container concurrency="10" connection-factory="connectionFactory">    
<jms:listener id="JdbcListener" destination="topicDest" ref="queueListener" />  
</jms:listener-container>

I found, that Camel and its idempotentConsumer suppose to solve duplication problem - of course, it would be nice to know why it happens in first place. The third question concerns Camel's configuration. I have this configuration (default):

<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="brokerURL" value="tcp://localhost:0"/>
</bean>

<bean id="myRepo" class="org.apache.camel.processor.idempotent.MemoryIdempotentRepository"/>

<camelContext xmlns="http://camel.apache.org/schema/spring">
<route>
    <from uri="direct:start"/>
    <idempotentConsumer messageIdRepositoryRef="myRepo">
        <header>messageId</header>
        <to uri="mock:result"/>
    </idempotentConsumer>
</route>
</camelContext>

Does it apply for all queues or should I make explicit subscription? I suppose it will check every topic/queue and all incoming messages. The problem at the moment, that all messages have messageId=null and the filter takes it as the parameter.

2011-03-01 11:24:09,152 DEBUG (org.springframework.jms.core.JmsTemplate:567) - Sending created message: ActiveMQMapMessage {commandId = 0, responseRequired = false, **messageId = null**, originalDestination = null, originalTransactionId = null, producerId = null, destination = null, transactionId = null, expiration = 0, timestamp = 0, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = false, type = null, priority = 0, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false} ActiveMQMapMessage{ theTable = {field=1, timestamp=1298975049138, price=72.89, tickerId=2} }

I didn't find a easy way of setting messageId. My question - is it enough to set messageId and it will work as excepted or something is wrong with configuration, for example I have to specify which topic will be used.

Thanks,

Dzidas

Answer 1

when using a JMS topic, you need to set the concurrent/max concurrent consumers to "1" or you will get duplicates. if you need multi-threaded consumption and/or load balancing, then use virtual topics instead.