Announcement Announcement Module
Collapse
No announcement yet.
Spring AMQP: would like to put message to queue and send ACK immediately Page Title Module
Move Remove Collapse
X
Conversation Detail Module
Collapse
  • Filter
  • Time
  • Show
Clear All
new posts

  • Spring AMQP: would like to put message to queue and send ACK immediately

    I wrote Java application, that sends messages to RabbitMQ. Then Flume picks messages up from RabbitMQ queue. I'm interested that nobody pulls messages from the queue, except flume.

    My application uses Spring AMQP Java plugin.
    The problem:

    With the code below, message comes to RabbitMQ queue and stays 'Unknowledges' for ever. As I understand, RabbitMQ is waiting for ACK from MessageListener, but MessageListener will never ACK. Does anybody have idea how to fix it?

    The code:

    public class MyAmqpConfiguration {

    @Autowired
    ConnectionFactory connectionFactory;

    @Bean
    public SimpleMessageListenerContainer messageListenerContainer() {

    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    container.setQueues(activityLogsQueue());
    container.setMessageListener(MyMessageListener());
    container.setConcurrentConsumers(3);

    return container;
    }

    @Bean(name="myTemplate")
    public RabbitTemplate rabbitTemplate() {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setMessageConverter(MyMessageConverter()) ;
    return template;
    }
    }


    public class MyMessageListener implements MessageListener {


    public MyMessageListener(MessageConverter converter, MyMessageHandler<MyObject> messageHandler) {
    this.converter = converter;
    this.messageHandler = messageHandler;
    }

    @Override
    public void onMessage(Message message) {
    this.messageHandler.doThings();
    }

    }

    public class MyMessageHandler {

    @Autowired
    @Qualifier("myTemplate")
    RabbitTemplate template;

    @Override
    public void handleMessage(MyObject thing) {
    template.convertAndSend(exchange, routingKey, thing);
    }

    }


    public class MyMessageConverter extends JsonMessageConverter {

    @Override
    protected Message createMessage(Object object, MessageProperties messageProperties) {
    //do things
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
    throw new UnsupportedOperationException("fromMessage is not supported in "+this.getClass().getName());
    }


    }

  • #2
    Your question is not clear.

    If Flume is the consumer, why do you have a listener?

    Senders don't ack messages, consumers do; the ack is the responsibility of the consumer.

    Comment


    • #3
      Hi,

      Thank you for the answer.
      Actually, Flume is consumer.
      Then, should I just leave template without listener?

      Originally posted by Gary Russell View Post
      Your question is not clear.

      If Flume is the consumer, why do you have a listener?

      Senders don't ack messages, consumers do; the ack is the responsibility of the consumer.

      Comment


      • #4
        Yes; if you only want to send; you only need a template, but I am interested in why you thought you needed a listener in the first place.

        However, this does mean your problem (missing acks) is a problem with your Flume configuration because, like I said, it's the consumer that acks.

        Comment


        • #5
          When I create template without MessageListener, I get following Exception:

          WARN org.springframework.amqp.rabbit.listener.SimpleMes sageListenerContainer - Consumer raised exception, processing can restart if the connection factory supports it. Exception summary: java.lang.IllegalStateException: No message listener specified - see property 'messageListener'

          Originally posted by Gary Russell View Post
          Yes; if you only want to send; you only need a template, but I am interested in why you thought you needed a listener in the first place.

          However, this does mean your problem (missing acks) is a problem with your Flume configuration because, like I said, it's the consumer that acks.

          Comment


          • #6
            Well, you'll get that if you define a listener Container and no listener, but you don't need a ListenerContainer.

            Comment


            • #7
              Perfect!
              I removed unnesessary ListeneConteiner, and it fixed the problem with 'unuknowledged' messages!!!
              Thank you!

              Originally posted by Gary Russell View Post
              Well, you'll get that if you define a listener Container and no listener, but you don't need a ListenerContainer.

              Comment

              Working...
              X