Блог

Я использую Spring Boot и Rappitmq, чтобы реализовать асинхронный обмен сообщениями между двумя разными приложениями. Это работает нормально, когда я обмениваюсь строками только в качестве содержимого сообщения. Я также реализовал обмен pojo. Из-за того, что pojos находятся в разных пакетах, процесс сериализации / десериализации завершается с ошибкой. Чтобы решить эту проблему, я использую JacksonConverter для использования json. Это работает, как я вижу в заголовке сообщения в службе подписки «application / json».

Модель для обмена (то же самое в службе публикации и подписчика):

 public class Employee{

private Integer id;
private String firstName;
private String lastName;
private Double salary;

public Employee(@JsonProperty("id") Integer id,
                @JsonProperty("firstName") String firstName,
                @JsonProperty("lastName") String lastName,
                @JsonProperty("salary") Double salary) {
    this.id = id;
    this.firstName = firstName;
    this.lastName = lastName;
    this.salary = salary;
}

public Integer getId() {
    return id;
}

public void setId(Integer id) {
    this.id = id;
}

public String getFirstName() {
    return firstName;
}

public void setFirstName(String firstName) {
    this.firstName = firstName;
}

public String getLastName() {
    return lastName;
}

public void setLastName(String lastName) {
    this.lastName = lastName;
}

public Double getSalary() {
    return salary;
}

public void setSalary(Double salary) {
    this.salary = salary;
}

@Override
public String toString() {
    return "{"  
            "id="   id  
            ", firstName='"   firstName   '''  
            ", lastName='"   lastName   '''  
            ", salary="   salary  
            '}';
}
  

Компоненты для использования Jackson2JsonConverter в службе publisher:

  @SpringBootApplication
public class PublisherApplication {

public static void main(String[] args) {

    SpringApplication.run(PublisherApplication.class, args);

}

@Bean
public Jackson2JsonMessageConverter producerMessageConverter(){
    return new Jackson2JsonMessageConverter();
}

@Bean
public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory){
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMessageConverter(producerMessageConverter());
    return rabbitTemplate;
}
  

}

Отправить сообщение:

     @Override
public void sendEmployeeObject(Employee employee) throws JsonProcessingException {
    rabbitTemplate.convertAndSend(RabbitMqConstants.EXCHANGE_NAME, "foo.bar.baz", employee);
}
  

}

Конфигурация подписчика:

    @Configuration
public class EmployeeMessageSubscriberConfig {

@Bean
@Qualifier("employeeQueue")
Queue queue() {
    return new Queue(RabbitMqConstants.EMPLOYEE_QUEUE, false);
}

@Bean
@Qualifier("employeeExchange")
TopicExchange exchange() {
    return new TopicExchange(RabbitMqConstants.EXCHANGE_NAME);
}

@Bean
@Qualifier("employeeQueueExchangeBinding")
Binding binding(Queue queue, TopicExchange exchange) {
    return BindingBuilder.bind(queue).to(exchange).with("foo.bar.#");
}
@Bean
public Jackson2JsonMessageConverter converter() {
    return new Jackson2JsonMessageConverter();
}
  

}

Метод прослушивателя:

  @RabbitListener(queues = RabbitMqConstants.EMPLOYEE_QUEUE)
    public void handleMessage(Employee employee) {

        System.out.println("Received <"   employee.toString()   ">");
    }
  

Чтобы проверить это, я отправляю объект Employee через rest api в службу publisher. Здесь я просто передаю его без изменений в rabbit. Служба подписчиков считывает сообщение и просто выводит его на консоль.

Проблема: Служба подписчиков печатает правильное значение:

 Received <{id=1, firstName='John', lastName='Doe', salary=1.5}>
  

но я также получаю трассировку ошибки:

     Caused by: com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot deserialize instance of `java.lang.String` out of START_OBJECT token
 at [Source: (String)"{"id":1,"firstName":"John","lastName":"Doe","salary":1.5}"; line: 1, column: 1]
    at com.fasterxml.jackson.databind.exc.MismatchedInputException.from(MismatchedInputException.java:59) 
  

Итак, насколько я понимаю, подписчик видит, что входящий Json является объектом типа ob, но он пытается десериализовать его в виде строки.

Вопрос: Почему выводится правильный вывод, когда трассировка говорит, что, похоже, произошла ошибка? Когда я использую RestController, входящий объект автоматически сопоставляется с заданной моделью. Должен ли я каким-то образом подключать это при использовании rabbitmq?

Редактировать:

Это руководство, которому я следовал.

Как я видел в заголовке моего подписчика, есть поле headers={__TypeId__=com.publisher.model.Employee} поэтому я установил модель Employee в обоих приложениях для пакета с одинаковым именем. Все та же ошибка. (Как я и ожидал, потому что в отличие от standardMessageConverter имя пакета не используется для сериализации)

Очевидно, что вы не можете получить сообщение с помощью java-компонента напрямую.

попробуйте использовать byte[] msg или String msg в качестве аргумента для вашего handleMessage() метода и преобразовать его в свой Employee компонент вручную.

Если вы хотите получить свое сообщение с помощью java-компонента, конфигурация поможет.

 @Configuration
public class RabbitMQConfig {
    
    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        return factory;
    }

}