RabbitMQ可靠性消息投递

RabbitMQ可靠性消息投递

已有父工程

Provider的CallBack

pom.xml

    
        RabbitMQ
        org.example
        1.0-SNAPSHOT
    
    4.0.0

    Providerr

    
        8
        8
    

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

        
            org.springframework.amqp
            spring-rabbit-test
            test
        

        
            org.springframework
            spring-web
        

        
        
            org.springframework.boot
            spring-boot-starter-web
        
    

启动类

在启动类中配置队列和交换机，最后将二者绑定

package com.example;


import com.rabbitmq.client.AMQP;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class RabbitProviderApplication {
    public static void main(String[] args) {
        SpringApplication.run(RabbitProviderApplication.class,args);
    }

    //创建队列
    public Queue createQueue(){
//        return new Queue("queue_demo01");
        return new Queue("item_queue");
    }

    //创建交换机
    @Bean
    public DirectExchange createExchange(){
        return new DirectExchange("exchange_direct_demo01");
    }

    //创建绑定
    @Bean
    public Binding createBinding(){
        return BindingBuilder.bind(createQueue()).to(createExchange()).with("item_insert");
    }
}

Provider的CallBack

application.xml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: /
    username: chocoMoss
    password: 123456
    publisher-/confirm/i-type=correlated: correlated
    publisher-return-type=correlated: correlated

server:
  port: 8080

My/confirm/iCallback

实现：RabbitTemplate./confirm/iCallback
重写方法

package com.example./confirm/i;

import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;

@Component
public class My/confirm/iCallback implements RabbitTemplate./confirm/iCallback {

    @Override
    public void /confirm/i(@Nullable CorrelationData correlationData, boolean ack, @Nullable String s) {
        if (ack){
            System.out.println("发送消息到交换机："+s);
        }else {
            System.out.println("发送消息到交换机失败，原因是："+s);
        }
    }
}

TestController

自动注入RabbitTemplate和其中的 /confirm/iCallback
将/confirm/iCallback设置进RabbitTemplate

package com.example.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/test")
public class TestController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitTemplate./confirm/iCallback /confirm/iCallback;

    //发送消息
    @RequestMapping("/send1")
    public String send1(){
        //设置回调函数
        rabbitTemplate.set/confirm/iCallback(/confirm/iCallback);

        //发送消息
        rabbitTemplate.convertAndSend("exchange_direct_demo01", "item.insert", "hello insert");

        return "ok";
    }
}

Provider的ReturnCallBack

MyReturnMessage

package com.example.returncallback;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

public class MyReturnMessage implements RabbitTemplate.ReturnCallback {
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("退回的消息是："+new String(message.getBody()));
        System.out.println("退回的replyCode是："+replyCode);
        System.out.println("退回的replyText是："+replyText);
        System.out.println("退回的exchange是："+exchange);
        System.out.println("退回的routingKey是："+routingKey);
    }
}

Controller

相应的，在controller中添加 ReturnCallback

package com.example.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/test")
public class TestController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitTemplate./confirm/iCallback /confirm/iCallback;

    @Autowired
    private RabbitTemplate.ReturnCallback returnCallback;

    //发送消息
    @RequestMapping("/send1")
    public String send1(){
        //设置回调函数
        rabbitTemplate.set/confirm/iCallback(/confirm/iCallback);

        //发送消息
        rabbitTemplate.convertAndSend("exchange_direct_demo01", "item.insert", "hello insert");

        return "ok";
    }

    //发送消息
    @RequestMapping("/send2")
    public String send2(){
        //设置回调函数
        rabbitTemplate.setReturnCallback(returnCallback);

        //发送消息
        rabbitTemplate.convertAndSend("exchange_direct_demo01", "item.insert", "hello insert");

        return "ok";
    }
}