Spring-RabbitMQ - mit manueller Kanal-Bestätigung auf einen Dienst mit @RabbitListener Konfiguration

Wie zu erkennen die Nachrichten manuell, ohne die Verwendung der automatischen Anerkennung.
Gibt es eine Möglichkeit, verwenden Sie diese zusammen mit der @RabbitListener und @EnableRabbit - Stil der Konfiguration.
Der größte Teil der Dokumentation erzählt uns SimpleMessageListenerContainer zusammen mit ChannelAwareMessageListener.
Jedoch mit, dass wir verlieren die Flexibilität, die bereitgestellt wird mit den Anmerkungen.
Ich konfiguriert haben, meinen Dienst als unten :

@Service
public class EventReceiver {

@Autowired
private MessageSender messageSender;

@RabbitListener(queues = "${eventqueue}")
public void receiveMessage(Order order) throws Exception {

  //code for processing order
}

Meine RabbitConfiguration ist als unten

@EnableRabbit
public class RabbitApplication implements RabbitListenerConfigurer {

public static void main(String[] args) {
    SpringApplication.run(RabbitApplication.class, args);
}

@Bean


public MappingJackson2MessageConverter jackson2Converter() {
        MappingJackson2MessageConverter converter = new  MappingJackson2MessageConverter();
        return converter;
    @Bean
public SimpleRabbitListenerContainerFactory myRabbitListenerContainerFactory() {
      SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
      factory.setConnectionFactory(rabbitConnectionFactory());
      factory.setMaxConcurrentConsumers(5);
      factory.setMessageConverter((MessageConverter) jackson2Converter());
      factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
      return factory;
    }

@Bean
public ConnectionFactory rabbitConnectionFactory() {
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
    connectionFactory.setHost("localhost");
    return connectionFactory;
}

@Override
public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
    registrar.setContainerFactory(myRabbitListenerContainerFactory());
}

@Autowired
private EventReceiver receiver;
}
}

Jede Hilfe wird sehr geschätzt, wie die Anpassung manuelle Kanal-Bestätigung zusammen mit den oben genannten Stil der Konfiguration.
Wenn wir implementieren die ChannelAwareMessageListener dann die onMessage-Signatur ändern.
Können wir implementieren ChannelAwareMessageListener auf einen Dienst ?

  • Die Frage, warum Sie selbst dies tun müssen. Wenn Ihr code ist wie in der Antwort unten (lehne auf Fehler, ansonsten ack), der container wird automatisch für Sie mit AUTO-ack-Modus - wenn der Hörer löst eine exception aus, wird die Nachricht abgelehnt werden; andernfalls acked.
InformationsquelleAutor Guru | 2016-08-02
  1. 21

    Fügen Sie die Channel zu den @RabbitListener Methode...

    @RabbitListener(queues = "${eventqueue}")
public void receiveMessage(Order order, Channel channel,
    @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
    ...
}

    und verwenden Sie das tag in der basicAck, basicReject.

    BEARBEITEN

    @SpringBootApplication
@EnableRabbit
public class So38728668Application {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = SpringApplication.run(So38728668Application.class, args);
        context.getBean(RabbitTemplate.class).convertAndSend("", "so38728668", "foo");
        context.getBean(Listener.class).latch.await(60, TimeUnit.SECONDS);
        context.close();
    }

    @Bean
    public Queue so38728668() {
        return new Queue("so38728668");
    }

    @Bean
    public Listener listener() {
        return new Listener();
    }

    public static class Listener {

        private final CountDownLatch latch = new CountDownLatch(1);

        @RabbitListener(queues = "so38728668")
        public void receive(String payload, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag)
                throws IOException {
            System.out.println(payload);
            channel.basicAck(tag, false);
            latch.countDown();
        }

    }

}

    Anwendung.Eigenschaften:

    spring.rabbitmq.listener.acknowledge-mode=manual
    • Danke für die Anregung und wir haben versucht, Ihre Anregungen und setzen Sie in der Zeile channel.basicAck('100001', false). Jetzt unabhängig davon, ob ich "true" oder "false" in der Zeile oben der code der listener und Warteschlange gehe in Endlosschleife. So können Sie mir helfen wie man das umgehen kann.
    • Wir haben endlich das Problem gelöst. Iam-Dokumentation dieses zum Wohle der anderen.
    • Ich bekam eine Fehlermeldung, dass der Frühling.rabbitmq.listener.acknowledge-Modus ist eine als veraltet markierte Eigenschaft. Ich landete Einstellung dieser Eigenschaft auf meine RabbitListenerContainerFactory Bean und habe es funktioniert auf diese Weise.
    • Das Anwesen wurde verschoben nach spring.rabbitmq.listener.simple.acknowledge-mode. Im Frühjahr Boot 2.0 kann es sein spring.rabbitmq.listener.simple.acknowledge-mode oder spring.rabbitmq.listener.direct.acknowledge-mode da Spring AMQP unterstützt jetzt 2 container-Typen. Siehe Dokumentation.
    • Könnte mir jemand erklären, warum java-Programmierer weigern, setzen die Einfuhren in Ihre code-Beispiele? Ich fühle, wie es mich retten würde Stunden.
    • Zu viel Lärm - wenn Sie eine IDE (z.B. eclipse/Intellij) es macht Vorschläge, und es ist normalerweise ziemlich offensichtlich, wenn es mehrere übereinstimmungen. In diesem bestimmten Beispiel ist der in meinem sandbox-github-repo.
    • Die info, wie die Anwendung verwenden, - Eigenschaften, um anzuerkennen-Modus auf "manuell" ist sehr wichtig! Man erwartet, zu konfigurieren, diese in die Parameter der @RabbitListener annotation.

    InformationsquelleAutor Gary Russell
  2. 4

    Nur für den Fall, Sie brauchen, um zu verwenden #onMessage() von ChannelAwareMessageListener Klasse. Dann können Sie es auf diese Weise.

    @Component
public class MyMessageListener implements ChannelAwareMessageListener {

@Override
public void onMessage(Message message, Channel channel) {
    log.info("Message received.");
    //do something with the message
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

    }

    Und für die rabbitConfiguration

    @Configuration
public class RabbitConfig {

public static final String topicExchangeName = "exchange1";

public static final String queueName = "queue1";

public static final String routingKey = "queue1.route.#";

@Bean
public ConnectionFactory connectionFactory() {
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
    connectionFactory.setUsername("xxxx");
    connectionFactory.setPassword("xxxxxxxxxx");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("vHost1");
    return connectionFactory;
}

@Bean
public RabbitTemplate rabbitTemplate() {
    return new RabbitTemplate(connectionFactory());
}

@Bean
Queue queue() {
    return new Queue(queueName, true);
}

@Bean
TopicExchange exchange() {
    return new TopicExchange(topicExchangeName);
}

@Bean
Binding binding(Queue queue, TopicExchange exchange) {
    return BindingBuilder.bind(queue).to(exchange).with(routingKey);
}


@Bean
public SimpleMessageListenerContainer listenerContainer(POCRabbitMessageListener pocRabbitMessageListener) {
    SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer();
    listenerContainer.setConnectionFactory(connectionFactory());
    listenerContainer.setQueueNames(queueName);
    listenerContainer.setMessageListener(pocRabbitMessageListener);
    listenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    listenerContainer.setConcurrency("4");
    listenerContainer.setPrefetchCount(20);
    return listenerContainer;
}

    }

    InformationsquelleAutor Pari Ngang
  3. -3

    Dank für gary ' s helfen. Ich endlich das Problem gelöst. Ich dokumentierte dies zum Wohle der anderen.
    Dies muss dokumentiert werden als Teil der standard-Dokumentation im Frühjahr AMQP Referenz-Dokumentation Seite.
    Service Klasse ist, wie unten beschrieben. 

       @Service
    public class Consumer {

    @RabbitListener(queues = "${eventqueue}")
    public void receiveMessage(Order order, Channel channel) throws Exception {



 //the above methodname can be anything but should have channel as second signature

    channel.basicConsume(eventQueue, false, channel.getDefaultConsumer()); 
    //Get the delivery tag
    long deliveryTag = channel.basicGet(eventQueue, false).getEnvelope().getDeliveryTag();
    try {

      //code for processing order

    catch(Exception) {
     //handle exception
        channel.basicReject(deliveryTag, true);
    }
    //If all logic is successful 
    channel.basicAck(deliveryTag, false);
}

    die Konfiguration wurde auch geändert, wie unten 

    public class RabbitApplication implements RabbitListenerConfigurer {

    private static final Logger log = LoggerFactory.getLogger(RabbitApplication .class);

    public static void main(String[] args) {
        SpringApplication.run(RabbitApplication.class, args);
    }

    @Bean
    public MappingJackson2MessageConverter jackson2Converter() {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        return converter;
    }

    @Bean
    public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setMessageConverter(jackson2Converter());
        return factory;
    }

    @Autowired
    private Consumer consumer;

    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        registrar.setMessageHandlerMethodFactory(myHandlerMethodFactory());
    }

      ...
}

    Hinweis: keine Notwendigkeit zu konfigurieren Rabbitconnectionfactory oder containerfactor etc, da die annotation implicity kümmern sich um alles.

    • NEIN - man sollte NICHT Thema einer basicConsume oder basicGet gegen den Kanal - basicGet Holen Sie eine weitere Nachricht. Die listener-container ist bereits das konsumieren von ihm und die Nachricht wird verwendet, um die Methode aufzurufen, hat eine andere Lieferung tag. Verwenden Sie stattdessen @Header(AmqpHeaders.DELIVERY_TAG) long tag. Siehe meine Antwort (edit).
    • Gary, entfernt basicConsume und basicGet und verwendet die @Header(AmqpHeaders.DELIVERY_TAG) langen tag in basicAck/ basicReject Der Fluss aufgehört zu arbeiten. Queue geholt wird, wieder und wieder in einer Endlosschleife und die Ziel-queue gefüllt ist. Rückgängig gemacht, den code zurück zu basicGet und basicConsume und es funktioniert.
    • Aber es funktioniert NICHT - Sie sind durch ACK bestätigt (und fallen) die nächste Nachricht.
    • also du meinst ich sollte benutzen eine Art zu konsumieren, es zu vermeiden. Es ist nicht klar, da Sie sagte vorhin, wir sollten nicht über basicConsume.
    • Die listener-container ist der Verbraucher - getDefaultConsumer() macht nichts in diesem Fall (zurück null) und basicGet holt die nächste Nachricht in der Warteschlange.
    • Ich schrieb eine schnelle Spring-Boot-app und es funktioniert genau so, wie ich beschrieben - bearbeitet ich meine Antwort mit dem code. Das komplette Projekt ist hier und der commit. Wenn Sie einen Haltepunkt auf basicAck können Sie sehen, die un-acked Nachricht in das Kaninchen Admin UI; Schritt über, und die Meldung ist acked.

    InformationsquelleAutor Guru
Schreibe einen Kommentar