阿裡雲AMQP Springboot內建

概述

本文主要介紹如何通過 Spring boot 連接配接阿裡雲AMQP服務。

操作步驟

1、示例程式下載下傳，下載下傳

位址

。

2、參數配置，

AMQP管理控制台

擷取。

• resources -> application.properties

spring.application.name=rabbitmq-demo

spring.rabbitmq.host=18********617278.mq-amqp.cn-hangzhou-a.aliyuncs.com
spring.rabbitmq.port=5672
spring.rabbitmq.username=******
spring.rabbitmq.password=******
spring.rabbitmq.virtual-host=******
spring.rabbitmq.template.mandatory=true
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true           

• RabbitConfig -> RESOURCE_OWNER_ID

private static final long RESOURCE_OWNER_ID =18********617278L;//資源owner賬戶 ID 資訊           

3、代碼調整（預設demo僅做了發送未被路由的消息的測試，為了測試相對完整，調整了部分代碼及注釋，友善進一步了解，也可以直接跳過此步驟直接運作測試即可）

• SenderWithCallback.class

package com.alibaba.rabbit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.UUID;


@Component
public class SenderWithCallback {
    Logger log= LoggerFactory.getLogger(SenderWithCallback.class);
    @Autowired
    private  RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void initRabbitTemplate() {
        // 設定生産者消息确認
        rabbitTemplate.setConfirmCallback(new RabbitConfirmCallback());
        rabbitTemplate.setReturnCallback(new RabbitReturnCallback());
    }

    public void send() {
        String exchange = "exchange-rabbit-springboot-advance5";
        String routingKey = "product";
        String unRoutingKey = "norProduct";

         //1.發送一條未被路由的消息  觸發returncallback、ConfirmCallback
        String message = LocalDateTime.now().toString() + "發送一條消息.";
        rabbitTemplate.convertAndSend(exchange, unRoutingKey, message, new CorrelationData("unRouting-" + UUID.randomUUID().toString()));
        log.info("發送一條消息,exchange:[{}],routingKey:[{}],message:[{}]", exchange, unRoutingKey, message);
        //2.發送一條路由的消息    觸發ConfirmCallback
        rabbitTemplate.convertAndSend(exchange, routingKey, message, new CorrelationData("Routing-" + UUID.randomUUID().toString()));
        log.info("發送一條消息,exchange:[{}],routingKey:[{}],message:[{}]", exchange, routingKey, message);
        //3.直接向queue中發送測試  供監聽消費測試
        rabbitTemplate.convertAndSend("queue", "test queue message.");
    }
}           

• RabbitReturnCallback.class

package com.alibaba.rabbit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * 設定 ReturnCallback 回調
 * 如果發送到交換器成功，但是沒有比對的隊列，就會觸發這個回調   在ConfirmCallback之前執行
 */
public class RabbitReturnCallback implements RabbitTemplate.ReturnCallback {

    Logger log= LoggerFactory.getLogger(RabbitReturnCallback.class);
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("message:"+message+",replyCode:"+replyCode+",replyText:"+replyText+",exchange:"+exchange+",routingKey:"+routingKey);
    }
}           

• RabbitConfirmCallback.class

package com.alibaba.rabbit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * 生産者端将消息發送出去，消息到達RabbitMQ之後，會傳回一個到達确認。
 * 這個确認實際上就是官方常說的ConfirmCallback，我們通過在生産者端使用一個回調類來監聽RabbiMQ傳回的消息确認。
 * Spring AMQP中我們通過設定RabbitTemplate的ConfirmCallback屬性來實作消息确認回調，通過一個實作了ConfirmCallback的類來實作回調邏輯。
 */
public class RabbitConfirmCallback implements RabbitTemplate.ConfirmCallback {

    Logger log= LoggerFactory.getLogger(RabbitConfirmCallback.class);
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.error("correlationData:"+correlationData+",ack:"+ack+",cause:"+cause);  //ack結果為true，表明正常接收到了消息
    }
}           

測試運作

• 啟動程式浏覽器請求send操作: http://localhost:8080/test/send
• 運作效果

019-06-07 12:42:43.693  INFO 2752 --- [nio-8080-exec-1] com.alibaba.rabbit.SenderWithCallback    : 發送一條消息,exchange:[exchange-rabbit-springboot-advance5],routingKey:[norProduct],message:[2019-06-07T12:42:43.612發送一條消息.]
2019-06-07 12:42:43.727 ERROR 2752 --- [124.156.22:5672] com.alibaba.rabbit.RabbitReturnCallback  : message:(Body:'2019-06-07T12:42:43.612發送一條消息.' MessageProperties [headers={spring_returned_message_correlation=unRouting-75d1580a-c93d-4d9c-bf06-2ed05ad8e464}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0]),replyCode:312,replyText:NO_ROUTE,exchange:exchange-rabbit-springboot-advance5,routingKey:norProduct
2019-06-07 12:42:43.728 ERROR 2752 --- [124.156.22:5672] c.alibaba.rabbit.RabbitConfirmCallback   : correlationData:CorrelationData [id=unRouting-75d1580a-c93d-4d9c-bf06-2ed05ad8e464],ack:true,cause:null
2019-06-07 12:42:43.753  INFO 2752 --- [nio-8080-exec-1] com.alibaba.rabbit.SenderWithCallback    : 發送一條消息,exchange:[exchange-rabbit-springboot-advance5],routingKey:[product],message:[2019-06-07T12:42:43.612發送一條消息.]
2019-06-07 12:42:43.891 ERROR 2752 --- [124.156.22:5672] c.alibaba.rabbit.RabbitConfirmCallback   : correlationData:CorrelationData [id=Routing-9ee37a35-0f8e-4a0d-b00b-2fc0fab574b4],ack:true,cause:null
2019-06-07 12:42:43.892 ERROR 2752 --- [124.156.22:5672] c.alibaba.rabbit.RabbitConfirmCallback   : correlationData:null,ack:true,cause:null
Receiver : test queue message.           

參考連結

amqp-demos rabbitmq生産者的消息确認