@RabbitListener 與 @RabbitHandler 及 消息序列化

        使用 @RabbitListener 注解標記方法，當監聽到隊列 debug 中有消息時則會進行接收並處理

　　使用 @Payload 和 @Headers 注解可以消息中的 body 與 headers 信息

@RabbitListener(queues = "debug")
public void processMessage1(@Payload String body, @Headers Map<String,Object> headers) {
    System.out.println("body："+body);
    System.out.println("Headers："+headers);
}

     可以獲取單個 Header 屬性

@RabbitListener(queues = "debug")
public void processMessage1(@Payload String body, @Header String token) {
    System.out.println("body："+body);
    System.out.println("token："+token);
}

　 通過 @RabbitListener 的 bindings 屬性聲明 Binding（若 RabbitMQ 中不存在該綁定所需要的 Queue、Exchange、RouteKey 則自動創建，若存在則拋出異常）

@RabbitListener(bindings = @QueueBinding(
        exchange = @Exchange(value = "topic.exchange",durable = "true",type = "topic"),
        value = @Queue(value = "consumer_queue",durable = "true"),
        key = "key.#"
))
public void processMessage1(Message message) {
    System.out.println(message);
}

• @RabbitListener 可以標注在類上面，需配合 @RabbitHandler 注解一起使用
• @RabbitListener 標注在類上面表示當有收到消息的時候，就交給 @RabbitHandler 的方法處理，具體使用哪個方法處理，根據 MessageConverter 轉換后的參數類型

@Component
@RabbitListener(queues = "consumer_queue")
public class Receiver {

    @RabbitHandler
    public void processMessage1(String message) {
        System.out.println(message);
    }

    @RabbitHandler
    public void processMessage2(byte[] message) {
        System.out.println(new String(message));
    }
    
}

MessageConvert

•  涉及網絡傳輸的應用序列化不可避免，發送端以某種規則將消息轉成 byte 數組進行發送，接收端則以約定的規則進行 byte[] 數組的解析

• RabbitMQ 的序列化是指 Message 的 body 屬性，即我們真正需要傳輸的內容，RabbitMQ 抽象出一個 MessageConvert 接口處理消息的序列化，其實現有 SimpleMessageConverter（默認）、Jackson2JsonMessageConverter 等
• 當調用了 convertAndSend 方法時會使用 MessageConvert 進行消息的序列化
• SimpleMessageConverter 對於要發送的消息體 body 為 byte[] 時不進行處理，如果是 String 則轉成字節數組,如果是 Java 對象，則使用 jdk 序列化將消息轉成字節數組，轉出來的結果較大，含class類名，類相應方法等信息。因此性能較差
• 當使用 RabbitMQ 作為中間件時，數據量比較大，此時就要考慮使用類似 Jackson2JsonMessageConverter 等序列化形式以此提高性能

　　消息處理方法參數是由 MessageConverter 轉化，若使用自定義 MessageConverter 則需要在 RabbitListenerContainerFactory 實例中去設置（默認 Spring 使用的實現是 SimpleRabbitListenerContainerFactory）

　　消息的 content_type 屬性表示消息 body 數據以什么數據格式存儲，接收消息除了使用 Message 對象接收消息（包含消息屬性等信息）之外，還可直接使用對應類型接收消息 body 內容，但若方法參數類型不正確會拋異常：

• application/octet-stream：二進制字節數組存儲，使用 byte[]
• application/x-java-serialized-object：java 對象序列化格式存儲，使用 Object、相應類型（反序列化時類型應該同包同名，否者會拋出找不到類異常）
• text/plain：文本數據類型存儲，使用 String
• application/json：JSON 格式，使用 Object、相應類型

使用 Java 序列化與反序列化

　　默認的 SimpleMessageConverter 在發送消息時會將對象序列化成字節數組，若要反序列化對象，需要自定義 MessageConverter

@Configuration
public class RabbitMQConfig {

    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new MessageConverter() {
            @Override
            public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
                return null;
            }

            @Override
            public Object fromMessage(Message message) throws MessageConversionException {
                try(ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(message.getBody()))){
                    return (User)ois.readObject();
                }catch (Exception e){
                    e.printStackTrace();
                    return null;
                }
            }
        });

        return factory;
    }

}

@Component
@RabbitListener(queues = "consumer_queue")
public class Receiver {

    @RabbitHandler
    public void processMessage1(User user) {
        System.out.println(user.getName());
    }

}

使用 JSON 序列化與反序列化

• RabbitMQ 提供Jackson2JsonMessageConverter 來支持消息內容 JSON 序列化與反序列化
• 消息發送者在發送消息時應設置 MessageConverter 為 Jackson2JsonMessageConverter

rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());

　　發送消息

User user = new User("jim");
rabbitTemplate.convertAndSend("topic.exchange","key.1",user);

　　消息消費者也應配置 MessageConverter 為 Jackson2JsonMessageConverter

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

}

　　消費消息

@Component
@RabbitListener(queues = "consumer_queue")
public class Receiver {

    @RabbitHandler
    public void processMessage1(@Payload User user) {
        System.out.println(user.getName());
    }

}

　　注意：被序列化對象應提供一個無參的構造函數，否則會拋出異常

 自定義監聽器DefaultConsumer 

public class MyConsumer extends DefaultConsumer {
    public MyConsumer(Channel channel){
        super(channel);
    }
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("-----------consume message----------");
        System.err.println("consumerTag: " + consumerTag);
        System.err.println("envelope: " + envelope);
        System.err.println("properties: " + properties);
        System.err.println("body: " + new String(body));
    }
}

 

參考：
　　https://www.jianshu.com/p/911d987b5f11