Modello richiesta-risposta utilizzando la libreria amqp di Spring

Question

tutti. Ho un'API HTTP per la pubblicazione di messaggi in un broker RabbitMQ e devo implementare il modello richiesta-risposta per ricevere le risposte dal server. Quindi sono qualcosa come un ponte tra i client e il server. Spingo i messaggi al broker con una chiave di routing specifica e c'è un Consumer per quei messaggi, che sta pubblicando i massaggi di ritorno come risposta e la mia API deve consumare la risposta per ogni richiesta. Così il diagramma è qualcosa di simile:

Quindi quello che faccio è la seguenti- Per ogni sessione HTTP creo un responseQueue temporaneo (che è legato allo scambio di default, con il routing chiave il nome di quel coda), dopo di che ho impostato l'intestazione replyTo del messaggio come nome della coda di risposta (dove aspetterò la risposta) e inoltre impostare il modello replyQueue su quella coda. Ecco il mio codice:

public void sendMessage(AbstractEvent objectToSend, final String routingKey) { 
    final Queue responseQueue = rabbitAdmin.declareQueue(); 
    byte[] messageAsBytes = null; 
    try { 
     messageAsBytes = new ObjectMapper().writeValueAsBytes(objectToSend); 
    } catch (JsonProcessingException e) { 
     e.printStackTrace(); 
    } 
    MessageProperties properties = new MessageProperties(); 
    properties.setHeader("ContentType", MessageBodyFormat.JSON); 
    properties.setReplyTo(responseQueue.getName()); 
    requestTemplate.setReplyQueue(responseQueue); 

    Message message = new Message(messageAsBytes, properties); 
    Message receivedMessage = (Message)requestTemplate.convertSendAndReceive(routingKey, message); 
} 

Allora, qual è il problema: Il messaggio viene inviato, dopo che si è consumato dal consumatore e la sua risposta è correttamente inviato alla coda destra, ma per qualche motivo non è preso di nuovo nel metodo convertSendAndReceived e dopo il timeout impostato my receivedMessage è nullo. Così ho provato a fare diverse cose - ho iniziato a ispezionare il codice sorgente (dal modo in cui è un vero incubo a farlo) e ho visto che non dichiaro la coda di risposta che crea un temporale per me, e l'intestazione replyTo è impostato sul nome della coda (lo stesso cosa faccio). Il risultato è stato lo stesso: il messaggio ricevuto è ancora nullo. Dopo di che ho deciso di utilizzare un altro modello che utilizza lo scambio di default, perché il responseQueue è legato a quello scambio:

requestTemplate.send(routingKey, message); 
Message receivedMessage = receivingTemplate.receive(responseQueue.getName()); 

Il risultato è stato la stesso-il responseMessage è ancora nulla. Le versioni di amqp e rabbit sono rispettivamente 1.2.1 e 1.2.0. Quindi sono sicuro che mi manca qualcosa, ma non so cosa sia, quindi se qualcuno mi può aiutare sarei estremamente grato.

Answer 1

1> E 'strano che RabbitTemplate utilizza doSendAndReceiveWithFixed se si fornisce il requestTemplate.setReplyQueue(responseQueue). Sembra che sia falso nella tua spiegazione.

2> Per rendere più lavorato con fisso ReplyQueue è necessario configurare un replyListenerContainer:

SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); 
container.setConnectionFactory(rabbitConnectionFactory); 
container.setQueues(responseQueue); 
container.setMessageListener(requestTemplate); 

3> Ma la parte più importante è intorno correlation. Il RabbitTemplate.sendAndReceive popola correlationId proprietà del messaggio, ma il lato consumatore deve ottenere accordo con esso, anche: non è sufficiente per inviare risposta alla responseQueue, il messaggio di risposta deve ha la stessa correlationId proprietà. Vedi qui: how to send response from consumer to producer to the particular request using Spring AMQP?

BTW non v'è alcuna ragione per popolare il Message manualmente: si può semplicemente sostenere Jackson2JsonMessageConverter al RabbitTemplate e sarà convertire il vostro objectToSend al JSON byte automaticamente con le intestazioni appropriate.