Duplicate Messages in ActiveMQ

Question

I use ActiveMQ as a JMS broker and consumer, jmsTemplate for sending messages, 1 is not a long-term topic at the moment. At peak time, I have ~ 100 messages per second.

It doesn't matter how many messages are in the queue, but I often get duplicate messages. The temporary solution that I came up with is setting the index in the table - at the moment, all messages are saved only in the database.

My first question is: why are messages duplicated if I indicated a non-simple topic and no answer is 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 about jmsContainer configuration. What is the difference between the code above and the code below? The above code gives me a topic as a subscriber, and the code below gives me a queue.

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

, Camel idempotentConsumer - , , . Camel. ( ):

<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>

? , / . , messageId = null, .

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} }

messageId. - messageId, - , , , .

,

Dzidas

Answer 1

JMS / "1" . / , .