title: Spring-Boot with RabbitMQ and TaskQueue / WorkerQueue
body: <p>I'm trying to use the <code>TaskQueue</code> with RabbitMQ, enhanced with a Deadletter Queue. I have a number of Spring Instances which all take Tasks via REST, put them to the RabbitQueue and a RabbitListener Method receives the Message. The receiverMethod check's if it can to the work. if so basicAck, otherwise basicNack.
  My target is that if the receiver says NACK, the Messages goes to the DeadletterQ, stays there for a given time (ttl) and rejoins the default RabbitQueue to become
  a new send attempt. In hope that any service have now time.
  That's the short version of my plan.</p>
  
  <p>I've read on rabbit's website about Deadlettering and TaskQueue. But all I can get is more or less a ping-pong or better a rocking behavior.</p>
  
  <p><strong>My Configuration looks like:</strong></p>
  
  <pre><code>    @Configuration
      public class BunnyConfiguration implements RabbitListenerConfigurer, DefaultLoggingInterface {
          private static final String DEADLETTER = "-deadletter";
          // config just taking properties from application.properties
          @Autowired
          private BunnyConfig bunnyConfig;
  
          // create the Exchange
          @Bean
          TopicExchange exchange() {
              getLogger().info("Register topic exchange {}", bunnyConfig.getAmpqTaskTopicExchangeName());
              return new TopicExchange(bunnyConfig.getAmpqTaskTopicExchangeName());
          }
  
          // Create the working queue
          @Bean
          @Profile("default")
          Queue queue() {
              getLogger().info("Register queue {}", bunnyConfig.getAmpqTaskQueueName());
              Map&lt;String, Object&gt; args = new HashMap&lt;&gt;();
              args.put("x-message-ttl", 600000);
              args.put("x-dead-letter-exchange", bunnyConfig.getAmpqTaskTopicExchangeName());
              args.put("x-dead-letter-routing-key", bunnyConfig.getAmpqTaskRouteKey() + DEADLETTER);
              return new Queue(bunnyConfig.getAmpqTaskQueueName(), true, false, false, args);
          }
  
          // Bind working Queue to Exchange
          @Bean
          @Profile("default")
          Binding binding() {
              getLogger().info("Binding queue {} and exchange {}", bunnyConfig.getAmpqTaskQueueName(),
                      bunnyConfig.getAmpqTaskTopicExchangeName());
              return BindingBuilder.bind(queue()).to(exchange()).with(bunnyConfig.getAmpqTaskRouteKey());
          }
  
  
  
          // Create Deadletter Queue
          @Bean
          @Profile("default")
          Queue deadLetterQueue() {
              getLogger().info("Register DLQ {}", bunnyConfig.getAmpqTaskQueueName() + DEADLETTER);
              Map&lt;String, Object&gt; args = new HashMap&lt;&gt;();
              args.put("x-message-ttl", 60000);
              args.put("x-dead-letter-exchange", bunnyConfig.getAmpqTaskTopicExchangeName());
              args.put("x-dead-letter-routing-key", bunnyConfig.getAmpqTaskRouteKey());
              return new Queue(bunnyConfig.getAmpqTaskQueueName() + DEADLETTER, true, false, false, args);
          }
  
          // Bind deadletter to exchange
          @Bean
          @Profile("default")
          Binding deadLetterBinding() {
              getLogger().info("DLQ Binding queue {} and exchange {}", bunnyConfig.getAmpqTaskQueueName() + DEADLETTER,
                      bunnyConfig.getAmpqTaskTopicExchangeName());
              return BindingBuilder.bind(deadLetterQueue()).to(exchange())
                      .with(bunnyConfig.getAmpqTaskRouteKey() + DEADLETTER);
          }
  
          @Bean
          public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
              getLogger().info("Creating Rabbit Template");
              final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
              rabbitTemplate.setMessageConverter(producerJackson2MessageConverter());
              return rabbitTemplate;
          }
  
          @Bean
          public Jackson2JsonMessageConverter producerJackson2MessageConverter() {
              getLogger().info("Creating Jackson2JsonMessageConverter");
              return new Jackson2JsonMessageConverter();
          }
  
          @Bean
          public MappingJackson2MessageConverter consumerJackson2MessageConverter() {
              getLogger().info("Creating MappingJackson2MessageConverter");
              return new MappingJackson2MessageConverter();
          }
  
          @Bean
          public DefaultMessageHandlerMethodFactory messageHandlerMethodFactory() {
              getLogger().info("Creating DefaultMessageHandlerMethodFactory");
              DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
              factory.setMessageConverter(consumerJackson2MessageConverter());
              return factory;
          }
  
          @Override
          public void configureRabbitListeners(final RabbitListenerEndpointRegistrar registrar) {
              getLogger().info("Configure RabbitListenerEndpointRegistrar");
              registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
          }
  
      }
  </code></pre>
  
  <p><strong>My abstract Rabbit Service</strong> which contains the basic send/receive operations. The <code>putTaskToQueue</code> is executed after a REST Call with the needed Data.</p>
  
  <pre><code>    @Service
      public abstract class AbstractRabbitService implements RabbitService {
  
          @Autowired
          private RabbitTemplate rabbitTemplate;
  
          @Autowired
          private BunnyConfig bunnyConfig;
  
  
          @Override
          public void initServices() {
              // currently not needed
          }
  
          @Override
          public void putTaskToQueue(final RabbitMessage entity, final String routeKeyChannel)
                  throws JsonProcessingException {
              final String routingKey = bunnyConfig.getAmpqTaskRouteKey() + routeKeyChannel;
              rabbitTemplate.convertAndSend(bunnyConfig.getAmpqTaskTopicExchangeName(), routingKey, entity);
          }
  
          //The first Parameter RabbitMessage is a simple serializeable Interface which nothing prescribed
          @RabbitListener(queues = "${bunny.ampqTaskQueueName}")
          public void receiveTask(final RabbitMessage message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag)
                  throws IOException {
  
              final boolean processed = processReceivedTask(message);
              if (processed) {
                  channel.basicAck(tag, false);
              } else {
                  channel.basicNack(tag, false, false);
              }
  
          }
  
          /**
          * 
          * 
          * @param message
          *            - the received object. You have to check the instance and take
          *            your own operations on/with it
          * @return - based on your return value the receiver method
          *         {@link AbstractRabbitService#receiveTask(RabbitMessage, Channel, long)}
          *         will ack or nack the message to the underlying queue
          */
          public abstract boolean processReceivedTask(final RabbitMessage message);
  
      }
  </code></pre>
  
  <p>My implementation of the <code>processReceivedTask</code> Method</p>
  
  <pre><code>    @Service
      public class RabbitServiceImpl extends AbstractRabbitService implements RabbitService {
  
          @Autowired
          private LocalConfig localConfig;
  
          @Autowired
          private UploadService uploadService;
  
          /**
          * @param message
          */
          @Override
          public boolean processReceivedTask(final RabbitMessage message) {
              if (message instanceof UploadTarget) {
                  final boolean maxParallsRunning = uploadService.isFull() || uploadService.isMaxParallelRunningTasksReached();
                  final UploadTarget received = (UploadTarget) message;
  
                  if (maxParallsRunning) {
                      if (logger.logger(this).isInfoEnabled()) {
                          logger.logger(this).info("Queue is full, rejecting {}", received.toTargetString());
                      }
                      return false;
                  }
  
                  if (logger.logger(this).isInfoEnabled()) {
                      logger.logger(this).info("Received new uploadtarget, try to add to upload queue");
                  }
  
                  return uploadService.addUploadTargetToQueue(received);
              }
              return false;
          }
  
      }
  </code></pre>
  
  <p>My Rabbit Settings</p>
  
  <pre><code>    #RabbitMQ Setup
      spring.rabbitmq.addresses=localhost
      spring.rabbitmq.host=localhost
      spring.rabbitmq.username=bunnyrabbit
      spring.rabbitmq.password=secrets
      spring.rabbitmq.port=5672
      spring.rabbitmq.listener.acknowledge-mode=manual
  
  
      bunny.ampqTaskTopicExchangeName=upload-dev-exchange
      bunny.ampqTaskQueueName=upload-dev
      bunny.ampqTaskRouteKey=upload.#
  </code></pre>
  
  <p>Normally the <code>uploadService.addUploadTargetToQueue(received)</code> function do some internal checks and put it in an internal <code>List&lt;UploadTarget&gt;</code> to do further work.
  Then return true. On any internal Errors it returns false. Quiet simple...</p>
  
  <p>If I start 2 Services with every have a max internal <code>List&lt;UploadTarget&gt;</code> Size about 5... It will never end. I add 20 Targets via REST. </p>
  
  <p>First 5 Message receives ok, than the internal List is full. The service "Nack" the RabbitMessage. And then goes it fast... After a few seconds my 
  DeadletterQ arrived >5000 Entries and the workerqueue about 10000... </p>
  
  <p>Step-by-Step (ohhh baby...)</p>
  
  <ol>
  <li><p>put new Message</p></li>
  <li><p>receive new Message<br>
  2.1 ack Message == deleted from queue<br>
  2.2 nack Message == put to deadletterQ to sleep a while<br>
      2.2.1 DeadletterQ<br>
      2.2.1.1 a Message reaches its TTL and get put back to WorkerQ.<br>
              It implies that it gets deleted from DeadletterQ  </p></li>
  </ol>
  
  <p>I don't even know if it would be feasible to solve my interpretation of the WorkerQueue or the Process. I think I have a missunderstanding of 
  the configuration part. But I don't know anymore.... Hope I can find here some more hints.</p>
  
  <p>Thanks for advice or some pushes to the right direction..</p>
  
  <p>-- Tom</p>