Как получить и ответить на весну

Question

Я пытаюсь развернуть RPC (шаблон запроса / ответа) и использую RabbitMQ и Spring на стороне сервера, потому что мне нужны динамические потребители. Я могу настроить динамических потребителей с помощью SimpleMessageListenerContainer, но я не знаю, как ответить на мое сообщение.

Вот моя конфигурация класса:

@Configuration
public class dynamicConsumerConfig {


    private static Properties prop = new Properties();


    public static void setPropValues() throws IOException {

        File configFile = new File("src/main/resources/config.properties");

        InputStream inStream = new FileInputStream(configFile.getAbsolutePath());

        prop.load(inStream);

    }


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


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



    @Bean
    public DirectExchange exchange1() {
        return new DirectExchange("pdfqueues");
    }

    @Bean
    public Binding slowBind(DirectExchange exchange, Queue slowQueue) {

        return  BindingBuilder.bind(slowQueue)
                .to(exchange)
                .with("slow");

    }


    @Bean
    public Binding fastBind(DirectExchange exchange, Queue fastQueue) {

        return  BindingBuilder.bind(fastQueue)
                .to(exchange)
                .with("fast");

    }


    @Bean
    public ConnectionFactory connect() throws IOException {


        setPropValues();


        CachingConnectionFactory connection = new CachingConnectionFactory();


        connection.setHost(prop.getProperty("HOST"));
        connection.setUsername(prop.getProperty("USER"));
        connection.setPassword(prop.getProperty("PASS"));
        connection.setPort(Integer.parseInt(prop.getProperty("PORT")));

        return  connection;


    }

    @Bean
    public SimpleMessageListenerContainer container1(ConnectionFactory connection) throws IOException {

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        setPropValues();
        container.setConnectionFactory(connection);
        container.setQueueNames("slowQueue");

        container.setMessageListener(firstListener());


        container.setMaxConcurrentConsumers(8);
        container.setConcurrentConsumers(1);
        container.setConsecutiveActiveTrigger(1);
        container.setConsecutiveIdleTrigger(1);
        container.setTxSize(1);          
        container.setPrefetchCount(1);

        return container;
    }


    @Bean
    public MessageListener firstListener()
    {
        return new MessageListener() {
            @Override
            public void onMessage(Message message) {

                PdfBoxService pdfboxservice = new PdfBoxService(prop.getProperty("tmpPath"),prop.getProperty("imagicPath"),prop.getProperty("resources"),
                                                prop.getProperty("tessdata"),prop.getProperty("languages"));



                String picture = new String(message.getBody(), StandardCharsets.UTF_8);


                List<ImagePair> lip = null;

                try {

                    lip = new ArrayList<ImagePair>();
                    lip.add(new ImagePair("JPG", picture));


                } catch (FileNotFoundException e1) {
                    // TODO Auto-generated catch block
                    e1.printStackTrace();
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }




                try {
                    ByteArrayOutputStream output= pdfboxservice.ImgToPdf(lip, false, false, false, 1, 1);




                } catch (IOException | InterruptedException | TransformerException | BadFieldValueException
                        | TesseractException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

            }
        };
    }

В функции firstListener() я получаю сообщение. В данном случае это картинка. Картинка конвертируется из JPG в PDF. PDF хранится в переменной output.

Мне нужно ответить на это output в другой очереди, но у меня нет инструментов для этого. Я думаю, что мой код плохой шаблон, но я не знаю, как сделать шаблон RPC с динамическими потребителями, используя SimpleMessageListenerContainer.

Answer 1

Используйте MessageListenerAdapter с методом POJO, который возвращает результат, вместо реализации MessageListener самостоятельно.

Начиная с версии 2.0, было предоставлено удобное FunctionalInterface:

@FunctionalInterface
public interface ReplyingMessageListener<T, R> {

    R handleMessage(T t);

}

Это облегчает удобную настройку адаптера с использованием Java 8 lamdas:

new MessageListenerAdapter((ReplyingMessageListener<String, String>) data -> {
    ...
    return result;
}));