注解位置辨析

隨著對(duì)消息隊(duì)列的應(yīng)用日益推廣，在分布式系統(tǒng)中的使用可以極大的降低對(duì)各個(gè)組件間的耦合度，從而提高組件的處理效率。因?yàn)橄㈥?duì)列的存在，可以使我們對(duì)任務(wù)進(jìn)行異步處理，這樣可以減少請(qǐng)求響應(yīng)時(shí)間和解耦。同時(shí)由于使用了消息隊(duì)列，只要保證消息格式不變，消息的發(fā)送方和接收方并不需要彼此聯(lián)系，也不需要受對(duì)方的影響，即解耦和。
所謂解耦，就是說A 系統(tǒng)產(chǎn)生一條數(shù)據(jù)，發(fā)送到 MQ 里面去，哪個(gè)系統(tǒng)需要數(shù)據(jù)自己去 MQ 里面消費(fèi)。如果新系統(tǒng)需要數(shù)據(jù)，直接從 MQ 里消費(fèi)即可；如果某個(gè)系統(tǒng)不需要這條數(shù)據(jù)了，就取消對(duì) MQ 消息的消費(fèi)即可。這樣下來，A 系統(tǒng)壓根兒不需要去考慮要給誰發(fā)送數(shù)據(jù)，不需要維護(hù)這個(gè)代碼，也不需要考慮人家是否調(diào)用成功、失敗超時(shí)等情況。

所謂異步，那么 A 系統(tǒng)連續(xù)發(fā)送 3 條消息到 MQ 隊(duì)列中，假如耗時(shí) 5ms，A 系統(tǒng)從接受一個(gè)請(qǐng)求到返回響應(yīng)給用戶，總時(shí)長(zhǎng)是 3 + 5 = 8ms，對(duì)于用戶而言，其實(shí)感覺上就是點(diǎn)個(gè)按鈕，8ms 以后就直接返回了，爽！網(wǎng)站做得真好，真快！

此外使用消息隊(duì)列還有削峰的優(yōu)勢(shì)。所謂削峰，即在某些時(shí)刻，用戶會(huì)大量的對(duì)我們的服務(wù)發(fā)起請(qǐng)求，我們的數(shù)據(jù)庫有時(shí)候需要對(duì)這些請(qǐng)求進(jìn)行寫入，但是呢，mysql的吞吐量頂破天就5000，剩下的就要慢慢等了，而且當(dāng)并發(fā)量過高的時(shí)候，數(shù)據(jù)庫的各種異常也會(huì)讓人抓狂，但是呢，我們使用消息隊(duì)列就不一樣了，用戶的各種請(qǐng)求通通塞入消息隊(duì)列里面，之后由消息隊(duì)列返回處理結(jié)果，而請(qǐng)求存儲(chǔ)在隊(duì)列里面，一個(gè)個(gè)按順序消費(fèi)，使請(qǐng)求寫入不出現(xiàn)高峰低谷

基于這些有點(diǎn)，我們開發(fā)團(tuán)隊(duì)最近在spring boot的開發(fā)過程中，由于項(xiàng)目的需要我們進(jìn)行消息隊(duì)列的接入改造。
在改造過程中遇到了這樣的問題，起初我將注解寫在了class上，在運(yùn)行的過程中會(huì)出現(xiàn)異常，以下是異常的詳細(xì)內(nèi)容：

2019-05-31 17:42:36.798 WARN 30544 --- [cTaskExecutor-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.

org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:1506)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1417)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1337)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1324)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1303)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:817)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:801)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700(SimpleMessageListenerContainer.java:77)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1042)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.springframework.amqp.AmqpException: No method found for class java.util.LinkedHashMap
at org.springframework.amqp.rabbit.listener.adapter.DelegatingInvocableHandler.getHandlerForPayload(DelegatingInvocableHandler.java:147)
at org.springframework.amqp.rabbit.listener.adapter.DelegatingInvocableHandler.getMethodNameFor(DelegatingInvocableHandler.java:250)
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.getMethodAsString(HandlerAdapter.java:70)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:190)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:120)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1414)
... 8 common frames omitted

至于說開發(fā)的源碼，我是這么寫的，我在class這里進(jìn)行注解，這個(gè)時(shí)候我猜測(cè)，應(yīng)該是注解的位置不對(duì)

@Component
@RabbitListener(queues = "xx.yy.zz")
public class Receiver {
    @RabbitHandler
    public void process(MSGSTO message) {
        System.out.println("消費(fèi)消息");
        System.out.println(message.toString());
    }
}

事實(shí)上，確實(shí)是位置不對(duì)，但更加專業(yè)的解答方式是，這個(gè)listener注解是方法級(jí)別上的，而不能用在class上面，我們不妨來看下RabbitListener的源碼，從根本上理解這個(gè)方法的使用。

package org.springframework.amqp.rabbit.annotation;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Repeatable;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

import org.springframework.messaging.handler.annotation.MessageMapping;
@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@MessageMapping
@Documented
@Repeatable(RabbitListeners.class)
public @interface RabbitListener {
   String id() default "";

   String containerFactory() default "";

   String[] queues() default {};

   Queue[] queuesToDeclare() default {};

   boolean exclusive() default false;

   String priority() default "";

   String admin() default "";

   QueueBinding[] bindings() default {};

   String group() default "";

   String returnExceptions() default "";

   String errorHandler() default "";

   String concurrency() default "";

   String autoStartup() default "";

}

由于業(yè)務(wù)需要，我們確實(shí)是需要對(duì)消息進(jìn)行異步處理，而異步接收消息的最簡(jiǎn)單的方法是使用帶注解的監(jiān)聽端點(diǎn)基礎(chǔ)結(jié)構(gòu)。簡(jiǎn)而言之，它允許將托管bean的方法公開為Rabbit listener的端點(diǎn)。在這里，使用queues屬性時(shí)，可以指定關(guān)聯(lián)的容器可以監(jiān)聽多個(gè)隊(duì)列?？梢允褂聾Header注釋來創(chuàng)建POJO方法可接收消息的隊(duì)列名稱。
br/>在這里，使用queues屬性時(shí)，可以指定關(guān)聯(lián)的容器可以監(jiān)聽多個(gè)隊(duì)列?？梢允褂聾Header注釋來創(chuàng)建POJO方法可接收消息的隊(duì)列名稱。

@Component
public class Receiver {

    @RabbitListener(queues = "xx.yy.zz")
    @RabbitHandler
    public void process(MSGSTO message) {
        System.out.println("消費(fèi)消息");
        System.out.println(message.toString());

至于說配置方式，我是通過application.yml的形式進(jìn)行接入配置的，例如

rabbitmq:
  addresses: 127.0.0.1
  port: 5672
  username: guest
  password: guest
  publisher-confirms: true
  publisher-returns: true
  virtual-host: dev
  listener:
      simple:
          concurrency: 10
          max-concurrency: 20

這些屬性會(huì)被注入到RabbitProperties屬性中，如

@ConfigurationProperties(prefix = "spring.rabbitmq")
public class RabbitProperties {
    …
}

挺有趣的對(duì)吧：）

參考資料：

1. 為什么使用消息隊(duì)列，https://www.javazhiyin.com/22897.html