Получайте и отправляйте объекты Java с помощью Spring AMQP

Question

Я хочу реализовать пример Spring AMQP для отправки и получения объектов Java с использованием слушателя.Я попробовал это:

Отправить объект Java

ConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
AmqpAdmin admin = new RabbitAdmin(connectionFactory);
admin.declareBinding(BindingBuilder.bind(new Queue(QUEUE_PROCESSING_TRANSACTION, false)).to(new TopicExchange(EXCHANGE_PROCESSING)).with(ROUTING_KEY_PROCESSING_TRANSACTION));              
AmqpTemplate template = new RabbitTemplate(connectionFactory);

TransactionsBean obj = new TransactionsBean();
obj.setId(Long.valueOf(111222333));

Получить и отправить обратно другой объект Java:

ConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
AmqpAdmin admin = new RabbitAdmin(connectionFactory);
admin.declareBinding(BindingBuilder.bind(new Queue(QUEUE_PROCESSING_TRANSACTION, false))
                .to(new TopicExchange(EXCHANGE_PROCESSING)).with(ROUTING_KEY_PROCESSING_TRANSACTION));
AmqpTemplate template = new RabbitTemplate(connectionFactory);

TransactionsBean obj = (TransactionsBean) template.receiveAndConvert(QUEUE_PROCESSING_TRANSACTION);
System.out.println(" !!!!!!! Received id " + obj.getTransaction_id());

SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueues(new Queue(QUEUE_PROCESSING_TRANSACTION, false));

container.setMessageListener(new MessageListener() {
  @Override
  public void onMessage(Message message) {
    // Receive here Java object and send back another object
  }
});

Можете ли вы показать мне, как расширить код без сложногоаннотации просто простых слушателей?

Answer 1

Самый простой способ - использовать @RabbitListener - что стало еще проще при использовании Spring Boot, поскольку он будет подключать инфраструктурные компоненты (шаблон, администратор и т. Д.).

@SpringBootApplication
public class So51009346Application {

    public static final String QUEUE_PROCESSING_TRANSACTION = "q1";

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

    @Bean
    public ApplicationRunner runner(RabbitTemplate template) {
        return args -> {
            ReplyObject reply = (ReplyObject) template.convertSendAndReceive("ex", "rk", new RequestObject());
            System.out.println(reply);
        };
    }

    @Bean
    public Queue queue() {
        return new Queue(QUEUE_PROCESSING_TRANSACTION);
    }

    @Bean
    public TopicExchange te() {
        return new TopicExchange("ex");
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(te()).with("rk");
    }

}

class RequestObject implements Serializable {

    private static final long serialVersionUID = 1L;

}

class ReplyObject implements Serializable {

    private static final long serialVersionUID = 1L;

}

@Component
class Listener {

    @RabbitListener(queues = So51009346Application.QUEUE_PROCESSING_TRANSACTION)
    public ReplyObject process(RequestObject ro) {
        return new ReplyObject();
    }

}

Если вы этого не сделаетеЕсли вы хотите использовать эту аннотацию по какой-то причине, вы можете подключить контейнер с помощью MessageListenerAdapter ...

@SpringBootApplication
public class So51009346Application {

    public static final String QUEUE_PROCESSING_TRANSACTION = "q1";

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

    @Bean
    public ApplicationRunner runner(RabbitTemplate template) {
        return args -> {
            ReplyObject reply = (ReplyObject) template.convertSendAndReceive("ex", "rk", new RequestObject());
            System.out.println(reply);
        };
    }

    @Bean
    public SimpleMessageListenerContainer container(ConnectionFactory cf, Listener listener) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
        container.setQueueNames(QUEUE_PROCESSING_TRANSACTION);
        container.setMessageListener(new MessageListenerAdapter(listener, "process"));
        return container;
    }

    @Bean
    public Queue queue() {
        return new Queue(QUEUE_PROCESSING_TRANSACTION);
    }

    @Bean
    public TopicExchange te() {
        return new TopicExchange("ex");
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(te()).with("rk");
    }

}

class RequestObject implements Serializable {

    private static final long serialVersionUID = 1L;

}

class ReplyObject implements Serializable {

    private static final long serialVersionUID = 1L;

}

@Component
class Listener {

    public ReplyObject process(RequestObject ro) {
        return new ReplyObject();
    }

}

Вы, конечно, можете подключить контейнер самостоятельно, как и в вашем вопросе, используя адаптер, но, как правило, лучше позволить Spring управлять им как @Bean, иначе вы пропустите некоторые функции (например, публикация событий при сбоях, пустой контейнер).Адаптер получает ссылку на ваш прослушиватель запросов / ответов и имя метода для вызова.