摘要：它還為具有偵聽器容器的消息驅(qū)動的提供支持。接收消息當存在基礎結構時，可以使用任何來注釋以創(chuàng)建偵聽器端點。默認情況下，如果禁用重試并且偵聽器拋出異常，則會無限期地重試傳遞。

Spring AMQP項目將核心Spring概念應用于基于AMQP的消息傳遞解決方案的開發(fā)。它提供了一個“模板”作為發(fā)送和接收消息的高級抽象。它還為具有“偵聽器容器”的消息驅(qū)動的POJO提供支持。這些庫有助于管理AMQP資源，同時促進依賴注入和聲明性配置的使用。在所有這些情況下，您將看到Spring Framework中與JMS支持的相似之處。

該項目由兩部分組成; spring-amqp是基礎抽象，spring-rabbit是RabbitMQ實現(xiàn)。

用于異步處理入站消息的偵聽器容器

RabbitTemplate用于發(fā)送和接收消息

RabbitAdmin用于自動聲明隊列，交換和綁定

高級消息隊列協(xié)議（AMQP）是面向消息的中間件的平臺中立的線級協(xié)議。Spring AMQP項目將核心Spring概念應用于基于AMQP的消息傳遞解決方案的開發(fā)。Spring Boot為通過RabbitMQ使用AMQP提供了一些便利，包括spring-boot-starter-amqp “Starter”。

RabbitMQ是一個基于AMQP協(xié)議的輕量級，可靠，可擴展且可移植的消息代理。Spring用于RabbitMQ通過AMQP協(xié)議進行通信。

RabbitMQ配置由外部配置屬性控制 spring.rabbitmq.*。例如，您可以在以下部分聲明以下部分 application.properties：

spring.rabbitmq.host = localhost
spring.rabbitmq.port = 5672
spring.rabbitmq.username = guest
spring.rabbitmq.password = guest

如果ConnectionNameStrategy上下文中存在bean，則它將自動用于命名由自動配置創(chuàng)建的連接ConnectionFactory。有關RabbitProperties更多支持的選項，請參閱 。

Spring的AmqpTemplate和AmqpAdmin被自動配置，您可以直接自動裝配它們變成自己的豆類，如下面的例子：

@Component
public class MyBean {

    private final AmqpAdmin amqpAdmin;
    private final AmqpTemplate amqpTemplate;

    @Autowired
    public MyBean(AmqpAdmin amqpAdmin,AmqpTemplate amqpTemplate){
        this.amqpAdmin = amqpAdmin;
        this.amqpTemplate = amqpTemplate;
    }

    //....
}

如有必要，任何org.springframework.amqp.core.Queue定義為bean的都會自動用于在RabbitMQ實例上聲明相應的隊列。

要重試操作，可以啟用重試AmqpTemplate（例如，在代理連接丟失的情況下）：

spring.rabbitmq.template.retry.enabled = true
spring.rabbitmq.template.retry.initial-interval = 2s

默認情況下禁用重試。您還可以RetryTemplate 通過聲明RabbitRetryTemplateCustomizerbean來以編程方式自定義。

當存在Rabbit基礎結構時，可以使用任何bean來注釋 @RabbitListener以創(chuàng)建偵聽器端點。如果RabbitListenerContainerFactory 未定義，SimpleRabbitListenerContainerFactory則會自動配置默認值，您可以使用該spring.rabbitmq.listener.type屬性切換到直接容器 。如果 定義了a MessageConverter或MessageRecovererbean，它將自動與默認工廠關聯(lián)。

以下示例組件在someQueue隊列上創(chuàng)建偵聽器端點：

@Component
public class MyBean {

    @RabbitListener(queues = "someQueue")
    public void processMessage(String content){
        //....
    }

}

如果您需要創(chuàng)建更多RabbitListenerContainerFactory實例，或者如果要覆蓋默認實例，則Spring Boot提供了一個 SimpleRabbitListenerContainerFactoryConfigurer和一個 DirectRabbitListenerContainerFactoryConfigurer可用于初始化一個 SimpleRabbitListenerContainerFactory和一個 DirectRabbitListenerContainerFactory的設置，其設置與自動配置使用的工廠相同。

例如，以下配置類公開了另一個使用特定的工廠MessageConverter：

@Configuration
public class RabbitConfiguration {

    ConnectionFactory connectionFactory;

    @Bean
    public SimpleRabbitListenerContainerFactory myFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer){
        SimpleRabbitListenerContainerFactory factory =
                new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory,connectionFactory);
        factory.setMessageConverter(myMessageConverter());
        return factory;
    }

    private MessageConverter myMessageConverter() {
        //..
        return null;
    }

}

然后您可以在任何@RabbitListener注釋方法中使用工廠，如下所示：

@Component
public class MyBean {

    @RabbitListener(queues = "someQueue",concurrency = "myFactory")
    public void processMessage(String content){
        //..
    }

}

您可以啟用重試來處理偵聽器拋出異常的情況。默認情況下，RejectAndDontRequeueRecoverer使用，但您可以定義MessageRecoverer 自己的。當重試耗盡時，如果代理配置了這樣做，則拒絕該消息并將其丟棄或路由到死信交換。默認情況下，禁用重試。您還可以RetryTemplate通過聲明RabbitRetryTemplateCustomizerbean來以編程方式自定義。

默認情況下，如果禁用重試并且偵聽器拋出異常，則會無限期地重試傳遞。您可以通過兩種方式修改此行為：將defaultRequeueRejected屬性設置 為false以便嘗試零重新傳遞，或拋出一個AmqpRejectAndDontRequeueException信號來拒絕該消息。后者是啟用重試并且達到最大傳遞嘗試次數(shù)時使用的機制。

參考地址：

spring-amqp
spring-amqp-samples
下文教程Github地址

    org.springframework.boot
    spring-boot-starter-amqp

鏈接參數(shù)

spring:
  rabbitmq:
    host: 192.168.192.132
    port: 5672
    username: guest
    password: guest
    template:
      retry:
        enabled: true # 代理鏈接丟失、啟用重試AmqpTemplate
        initial-interval: 2s # 代理鏈接丟失、啟用重試AmqpTemplate

amqpAdmin可以用于交換器，隊列以及綁定的申明，以下代碼演示了amqpAdmin的主要用法。

public class MyAdmin extends SampleApplicationTests {

    @Autowired
    AmqpAdmin amqpAdmin;

    @Test
    public void contextLoads(){
        //新建交換器
        amqpAdmin.declareExchange(new DirectExchange("myself.directExchange"));
        //新建隊列
        amqpAdmin.declareQueue(new Queue("myself.queue",true));
        //綁定
        amqpAdmin.declareBinding(new Binding("myself.queue",Binding.DestinationType.QUEUE,"myself.directExchange","",null));
    }

    @Test
    public void remove(){
        //移除交換器
        amqpAdmin.deleteExchange("myself.directExchange");
        //移除隊列
        amqpAdmin.deleteQueue("myself.queue");
    }
}

快速使用的發(fā)送接收模板

public class MyTemplate extends SampleApplicationTests {

    @Autowired
    AmqpTemplate amqpTemplate;

    @Test
    public void send(){
        amqpTemplate.convertAndSend("myself.queue","nihao");
    }

}

@Service
public class MyService {

    @RabbitListener(queues = "myself.queue")
    public void recevie(String result){
        System.out.println("監(jiān)聽到消息了");
        System.out.println(result);
    }

}