out of sync problem?

[Michal Glowacki]

I have constructed a queue, to put incoming operations to be computed. So, when operation comes via onMessage method, it is put at end of ConcurrentLinkedQueue (was LinkedList earlier, but also was not working). After adding, I start (just once, after deploy) processing thread, which in interval is checking the list for operations to process. The problem is, that once per 2-3 days something goes wrong. I noticed, that operations are added correctly to the queue, however processing thread, sees the queue empty all the time. From the logs, there is no error. Could it be my queue is out of sync?


Here's the code

@MessageDriven(activationConfig = {
		@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
		@ActivationConfigProperty(propertyName = "destination", propertyValue = "/queue/ProcessingQueue"),
		@ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge")})
public class Processor implements MessageListener {

	Logger log = Logger.getLogger(Processor.class);

	private final static long OPERATION_LIST_CHECK_INTERVAL = 5000L;

	// private LinkedList<OperationModel > list =new  // I used that one in the past, also not working well
	// LinkedList<OperationModel >();
	ConcurrentLinkedQueue<OperationModel > list = new ConcurrentLinkedQueue<OperationModel >();
	public static Thread processingThread = null;

	@Override
	public void onMessage(Message message) {
		OperationModel grm = null;
		try {
			grm = (OperationModel ) ((ObjectMessage) message).getObject();
			log.info(grm.toString());
		} catch (JMSException ex) {
			log.error(
					"Processor.onMessage(): Error reading operation from queue",
					ex);
			return;
		}
				
		list.add(grm);

		log.info(processingThread);
		if (processingThread == null || !processingThread.isAlive()) {
			log.info("Initiating processing thread...");
			initProcessingThread();
			log.info("done");
		}
	}

	private void initProcessingThread() {
		processingThread = new Thread(new ProcessList());
		processingThread.setDaemon(true);
		processingThread.start();
	}

	class ProcessList implements Runnable {

		@SuppressWarnings("static-access")
		@Override
		public void run() {
			while (true) {
				try {
					log.info("Checking list");
					OperationModel operationModel = list.element();
					boolean result = processOperation(operationModel );				
					if (!result) {
						log.info("Removed " + list.poll().getOperationId());
					}

				} catch (NoSuchElementException ex) {
					try {
						log.info("List is empty, list size " + list.size());
						Thread.currentThread()
								.sleep(OPERATION_LIST_CHECK_INTERVAL );
					} catch (InterruptedException ex1) {
						log.fatal(
								"Processor.run(): Operation processing "
										+ "thread has been interrupted", ex1);
					}
					continue;
				}
			}
		}
}

[Christophe Verré]

The thing puzzling me is : what is that queue doing in a message-driven bean ?

[Michal Glowacki]

The processing is very complex and I don't want situation when two or more operations are processed the same time. That's why I put a queue there and a thread for processing one by one. I used MDB so the operations are processed asynchronously, so my web app does not have to wait for the result.