How to make sure a message is delivered no more than once

[radix_zero]

I have a queue set up with two clients listening to the queue. (non-persisent, auto acknowlege, JBoss 4.0.2) both clients are getting all of the messages sent to the queue. Essentially my queue is behaving in pub/sub.

how do I unsure that a message in a queue is delivered no more than once?

[darranl]

Are you sure that you are using a queue and not a topic?

[radix_zero]

yes, it is a Queue. I am a little suprised at this problem, although I have never done a Queue before usually always dealing with topics.

I need to distribute a process. a queue with multiple consumers seemed the perfect answer, because each task passed along could get grabbed by exactly one consumer.

 <mbean code="org.jboss.mq.server.jmx.Queue"
 name="jboss.mq.destination:service=Queue,name=uncategorizedpages">
 <depends optional-attribute-name="DestinationManager">jboss.mq:service=DestinationManager</depends>
 </mbean>

The code uses Queue in the usual way:

 Context ctx = new InitialContext(config);
 QueueConnectionFactory queueFactory = (QueueConnectionFactory) ctx.lookup("ConnectionFactory");
 Destination queueDestination = (Destination) ctx.lookup(unitQueueName);

 QueueConnection queueConnection = (QueueConnection) queueFactory.createConnection(unitQueueUser, unitPagesQueuePass);
 Session queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

 List<UnitVO> batch = loadBatch(0);


 // for each page in the batch
 for (UnitVO unit : batch)
 {
 MessageProducer producer = queueSession.createProducer(queueDestination);
 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
 ObjectMessage msg = queueSession.createObjectMessage(page);
 producer.send(msg);
 producer.close();

 }
 queueSession.close();
 queueConnection.close();

 Context ctx = new InitialContext(this.config);
 QueueConnectionFactory queueConnnectionFactory = (QueueConnectionFactory) ctx.lookup(connectionFactoryJNDIName);
 QueueConnection queueConnection = queueConnnectionFactory.createQueueConnection(unitQueueUser, unitQueuePass);
 QueueSession unitSession =
 queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
 Queue unitQueue =
 (Queue) ctx.lookup(unitQueueName);
 QueueReceiver receiver =
 unitSession.createReceiver(unitQueue);
 queueConnection.start();


 // until we are told to stop
 while (!shutdown)
 {
 Message msg = receiver.receive(1000);

 if (msg != null && msg instanceof ObjectMessage)
 {
 UnitVO page = (UnitVO) ((ObjectMessage) msg).getObject();
 System.out.println(unit.getId());
 }
 else {
 shutdown=true;
 }
 }

[radix_zero]

As I experiment with this more I find that it is a time delay in removing the message from the queue.

If i sleep the consumer thread for a small random interval it helps break things up. But I still get duplicates on each consumer.