rabbitmq整合springboot回调确认一体化

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

spring.application.name=spirng-boot-rabbitmq-sender
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=1qazxsw@
server.port=80
spring.rabbitmq.publisher-cOnfirms=true
spring.rabbitmq.publisher-returns = true
spring.rabbitmq.listener.direct.acknowledge-mode=manual
spring.rabbitmq.listener.simple.acknowledge-mode=manual

package com.rabbit.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class SenderConfig {
    
    public static final String EXCHANGE   = "exchange";
    public static final String ROUTINGKEY = "routingkey";
    public static final String QUEUE = "message";
    public static final String QUEUE1      = "message1";
        @Bean
        public Queue queueMessage() {
            return new Queue(QUEUE);
        }

        @Bean
        public Queue queueMessage1() {
            return new Queue(QUEUE1);
        }

        @Bean
        public TopicExchange exchange() {
            return new TopicExchange(EXCHANGE);
        }

        @Bean
        Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessage).to(exchange).with(QUEUE);
        }

        @Bean
        Binding bindingExchangeMessages(Queue queueMessage1, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessage1).to(exchange).with(QUEUE1);
        }
}

package com.rabbit.service;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.rabbit.config.SenderConfig;
import com.rabbit.model.User;

@Component
public class HelloSender implements RabbitTemplate.ReturnCallback{
    /*@Autowired
    private AmqpTemplate template;*/
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void send() {
        User user = new User();
        user.setUsername("测试");
        user.setPassword("12345");
        this.rabbitTemplate.setReturnCallback(this);
        this.rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                System.out.println("HelloSender消息发送失败" + cause + correlationData.toString());
            } else {
                System.out.println("HelloSender 消息发送成功 ");
            }
        });
        this.rabbitTemplate.convertAndSend(SenderConfig.EXCHANGE, SenderConfig.QUEUE1, user.getUsername());
    //template.convertAndSend("exchange", "topic.message", user.getUsername());
    }
   /* public void send1() {
        User user = new User();
        user.setUsername("测试");
        user.setPassword("12345");
    template.convertAndSend("exchange", "topic.message1", user.getPassword());
    }*/

    @Override
    public void returnedMessage(Message message, int replycode, String replytext, String exchange, String routingKey) {
        System.out.println("消息主题 message:"+message);
        System.out.println("消息主题 message:"+replycode);
        System.out.println("描述:"+replytext);
        System.out.println("消息使用的交换器 exchange:"+exchange);
        System.out.println("消息使用的路由器 routing:"+routingKey);
    }
    /*@Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("消息唯一标识" + correlationData.getId());
        System.out.println("确认结果" + ack);
        System.out.println("失败原因" + cause);
    }*/
}

package com.rabbit.config;

import java.io.IOException;
import java.util.Date;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

@Component
public class HelloReceive {

    @RabbitListener(queues=ReceiveConf.QUEUE)    //监听器监听指定的Queue
    public void process1(String str,Message message, Channel channel) throws IOException {  
        System.out.println("HelloReceiver收到  : " + str +"收到时间"+new Date());
        try {
            //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            System.out.println("receiver success");
        } catch (IOException e) {
            e.printStackTrace();
            //ack返回false，并重新回到队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
            System.out.println("receiver fail");
        }
    }
    @RabbitListener(queues=ReceiveConf.QUEUE1)    //监听器监听指定的Queue
    public void process2(String str) {
        System.out.println("message1:"+str);
    }

}