一、三個(gè)注解

• @EnableRabbit
• @RabbitListener
• @RabbitHandler

@EnableRabbit

@EnableRabbit和@Configuration一起使用，可以加在類或者方法上，這個(gè)注解開啟了容器對注冊的bean的@RabbitListener檢查。

@Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Documented @Import(RabbitBootstrapConfiguration.class) public @interface EnableRabbit { }

從源代碼上可以看出這個(gè)注解提供了很簡單的功能就是引入了另一個(gè)配置類：RabbitBootstrapConfiguration。該類注冊了兩個(gè)bean：一個(gè)BeanPostProcessor一個(gè)ListenerEndpointRegistry。

注冊的BeanPostProcessor則會在bean初始化之后掃描@RabbitListener和@RabbitHandler注解。

@Configuration public class RabbitBootstrapConfiguration {@Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)@Role(BeanDefinition.ROLE_INFRASTRUCTURE)public RabbitListenerAnnotationBeanPostProcessor rabbitListenerAnnotationProcessor() {return new RabbitListenerAnnotationBeanPostProcessor();}@Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)public RabbitListenerEndpointRegistry defaultRabbitListenerEndpointRegistry() {return new RabbitListenerEndpointRegistry();}}

@RabbitListener

@RabbitListener用于注冊Listener時(shí)使用的信息(注解聲明 Binding)：如queue，exchange，key、ListenerContainerFactory和RabbitAdmin的bean name。使用 @RabbitListener 注解標(biāo)記方法，當(dāng)監(jiān)聽到隊(duì)列 debug 中有消息時(shí)則會進(jìn)行接收并處理

@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); }

掃描到bean帶有該注解后，首先會將注解的內(nèi)容封裝到Endpoint對象中并和ListenerContainerFactory的實(shí)例一起添加到上面的RabbitListenerEndpointRegistry實(shí)例中。添加的時(shí)候會創(chuàng)建相應(yīng)的ListenerContainer實(shí)例并添加Listener對象。

RabbitListenerAnnotationBeanPostProcessor通過RabbitListenerEndpointRegistrar間接持有RabbitListenerEndpointRegistry實(shí)例。

@RabbitHandler

@RabbitListener 和 @RabbitHandler結(jié)合使用，不同類型的消息使用不同的方法來處理。@RabbitListener 可以標(biāo)注在類上面，需配合 @RabbitHandler 注解一起使用，@RabbitListener 標(biāo)注在類上面表示當(dāng)有收到消息的時(shí)候，就交給 @RabbitHandler 的方法處理，具體使用哪個(gè)方法處理，根據(jù) MessageConverter 轉(zhuǎn)換后的參數(shù)類型

@Component @RabbitListener(queuesToDeclare = @Queue(value = "hello")) public class HelloCustomer {@RabbitHandlerpublic void receive1(String message){System.out.println("message = " + message);}@RabbitHandlerpublic void receive2(byte[] message) {System.out.println(new String(message));} }

@Payload 與 @Headers

• 使用 @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); }

• 也可以獲取單個(gè) Header 屬性

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

二、Message 內(nèi)容對象序列化與反序列化

• 使用 Java 序列化與反序列化
  默認(rèn)的 SimpleMessageConverter 在發(fā)送消息時(shí)會將對象序列化成字節(jié)數(shù)組，若要反序列化對象，需要自定義 MessageConverter

@Configuration public class RabbitMQConfig {@Beanpublic RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setMessageConverter(new MessageConverter() {@Overridepublic Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {return null;}@Overridepublic 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 {@RabbitHandlerpublic void processMessage1(User user) {System.out.println(user.getName());}}

• 使用 JSON 序列化與反序列化

RabbitMQ 提供了 Jackson2JsonMessageConverter 來支持消息內(nèi)容 JSON 序列化與反序列化消息發(fā)送者在發(fā)送消息時(shí)應(yīng)設(shè)置 MessageConverter 為 Jackson2JsonMessageConverter

rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());

發(fā)送消息

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

消息消費(fèi)者也應(yīng)配置 MessageConverter 為 Jackson2JsonMessageConverter

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

消費(fèi)消息

@Component @RabbitListener(queues = "consumer_queue") public class Receiver {@RabbitHandlerpublic void processMessage1(@Payload User user) {System.out.println(user.getName());} }

參考文章

與50位技術(shù)專家面對面20年技術(shù)見證，附贈技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的Spring boot使用Rabbitmq注解及消息序列化的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。