package com.ya.publisher;

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

/**
 * @program: untitled
 * @description:
 * @author: ChestnutDuck
 * @create: 2023-12-26 21:28
 **/
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void testSendMessageQueue(){
        String queueName="demo.queue";
        String msg="hello ，我是publisher";
        this.rabbitTemplate.convertAndSend(queueName,msg);
    }
}

package com.ya.consumer.listener;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @program: untitled
 * @description:
 * @author: ChestnutDuck
 * @create: 2023-12-27 14:38
 **/
@Slf4j
@Component
public class MqListener {
    @RabbitListener(queues = "demo.queue")
    public void ListenSimpleQueue(String msg){
        System.out.println("收到了发送者的消息："+msg);
    }
}

@RabbitListener(queues = "work.queue")
    public void ListenWorkQueue(String msg){
        System.err.println("消费者1收到了发送者的消息："+msg); //err表示错误 打印出来的是红色的字
    }
    @RabbitListener(queues = "work.queue")
    public void ListenWorkQueue2(String msg){
        System.out.println("消费者2收到了发送者的消息："+msg);

    }

@Test
   void testWorkQueue() throws InterruptedException {
       String queueName="work.queue";
       for (int i=0;i<50;i++){
           String msg="hello ，我是publisher"+i;
           this.rabbitTemplate.convertAndSend(queueName,msg);
           Thread.sleep(20);
       }
   }

@Test
  void testSendFanout(){
      String exchange="ya.fanout"; //自己创建的fanout
      String msg="hello ，everyone";
      this.rabbitTemplate.convertAndSend(exchange,null,msg);
  }

@Test
void testSendDirect(){
    String exchange="ya.direct";
    String msg="红色警报，";
    this.rabbitTemplate.convertAndSend(exchange,"yellow",msg); //会根据你提供的不同的RoutingKey 去决定发送给内个消费者
}

@Test
  void testSendTopic(){
      String exchange="ya.topic";
      String msg="红色警报，";
      this.rabbitTemplate.convertAndSend(exchange,"china.news",msg); //这个只会绑定你设置好的通配符
  }

package com.ya.consumer.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @program: untitled
 * @description:
 * @author: ChestnutDuck
 * @create: 2023-12-27 16:19
 **/
@Configuration
public class FanoutConfiguration {
    @Bean
    public FanoutExchange fanoutExchange(){
        FanoutExchange build = (FanoutExchange) ExchangeBuilder.fanoutExchange("ya.fanout").build();
//        return new FanoutExchange("ya.fanout")
        return build;
    }
    @Bean
    public Queue FanoutQueue(){
//        QueueBuilder.durable("").build();
        return new Queue("fanout.queue");
    }
    @Bean
    public Binding fanoutBinding(){
        return BindingBuilder.bind(FanoutQueue()).to(fanoutExchange());
        //不是调用的方法，是在容器中找bean
    }

}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name="direct.queue1",durable = "true"),//是否可持续化
        exchange = @Exchange(name = "ya.direcct",type= ExchangeTypes.DIRECT),
        key = {"red","blue"}
))
public void ListenDirectQueue(String msg) throws InterruptedException {
    System.err.println("消费者1收到了发送者的消息："+msg);

}

    @Bean
    public MessageConverter jacksonMessageConvertor(){
        return new Jackson2JsonMessageConverter();
    }
//注意消费端和发送端都要注入这个bean

@Slf4j
@Configuration
public class ConfirmConfig implements ApplicationContextAware {
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        //获RabbitTemplate
        RabbitTemplate rabbitTemplate =applicationContext.getBean(RabbitTemplate.class);//设置Returncallback
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                log.debug("收到的消息的return callback,exchange:{},key:{},msg:{},code:{},text:{}",
                        returnedMessage.getExchange(),returnedMessage.getRoutingKey(),returnedMessage.getMessage(),
                        returnedMessage.getReplyCode(),returnedMessage.getReplyText());
            }
        });
    }
}

@Test
   void testConfirmCallBack(){
       //1. 创建cd
       CorrelationData cd = new CorrelationData(UUID.randomUUID().toString()); //获取一个随机的id
       //2. 添加ConfirmCallback
       cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>(){

           @Override
           public void onSuccess(CorrelationData.Confirm result) {
               if(result.isAck()){ // result.isAck() , boolean类型， true代表ack回执,false 代表nack回执
                   log.debug("发送消息成功，收到ack!");
               }else {// result.getReason(), string类型，返回nack时的异常描述
                   log.error("发送消息失败，收到 nack,reason : {}",result.getReason()); //之后应该是重发消息
               }
           }

           @Override
           public void onFailure(Throwable ex) {
               log.error("消息回调失败",ex);
           }
       });
       this.rabbitTemplate.convertAndSend("ya.direct","yellow","hello",cd);
        Thread.sleep(2000); //因为回调是需要时间 ，但是测试在单元测试中 单元测试结束后就销毁
   }

@Test
 void testPageOut(){
     Message build = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8)) //手动设置消息的字符集
             .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build(); //设置消息为非持久化消息
     for (int i=0;i<1000000;i++){
         this.rabbitTemplate.convertAndSend("fanout.queue",build);
     }

 }

@Bean
public Queue FanoutQueue(){
    QueueBuilder.durable("")
            .lazy() //开启Lazy模式
            .build();

    return new Queue("fanout.queue");
}

@RabbitListener(queuesToDeclare = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode",value = "lazy")
))
public void ListenLazyQueue(String msg)  {
    System.err.println("消费者1收到了发送者的消息："+msg);
}

@Configuration
@ConditionalOnProperty(prefix = "spring.rabbitmq.listener.simple.retry",
        name = "enabled",havingValue = "true") //只有这个spring.rabbitmq.listener.simple.retry.enabled=true 时，这个bean才会生效
public class ErrorConfiguration {
    @Bean
    public DirectExchange errorExchange(){
        return new DirectExchange("error.direct");
    }
    @Bean
    public Queue errorQueue(){
        return  new Queue("error.queue");
    }
    @Bean
    public Binding errorBinding(Queue errorQueue,DirectExchange errorExchange){
        return BindingBuilder.bind(errorQueue).to(errorExchange).with("error");
    }
}

@Bean
public MessageRecoverer ReoublishMessageRecoverer(RabbitTemplate rabbitTemplate){
    return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","erros"); //后面的时key
}

@Bean
public MessageConverter messageConverter(){
    Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();//2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
     jjmc.setCreateMessageIds(true) ;
    return jjmc;
}

    @Test
    void testSend(){
//        Message build = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8))
//                .setExpiration("10000")
//                .build();//这里不用消息转换器是因为上面的错误消息会发到之前的错误交换机中，这里silmle.queue没有绑定消费者mq会当成错误信息
           this.rabbitTemplate.convertAndSend("simple.direct", "hi", "hello", new MessagePostProcessor() {//这里是后置消息处理器
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setExpiration("3000");
                    return message;
                }
            });
            log.info("消息发送成功");
    }
}

@RabbitListener(bindings = @QueueBinding(
         value = @Queue(name="delay.queue",durable = "true"),//是否可持续化
         exchange = @Exchange(name = "ya.direct",type= ExchangeTypes.DIRECT,delayed="true" ),
         key = "delay"
 ))

   @Bean
    public FanoutExchange fanoutExchange(){
        FanoutExchange build = (FanoutExchange) ExchangeBuilder
                .fanoutExchange("delay.fanout")
                .delayed()  //设置delayed属性为true
                .durable(true) //持久化
                .build();
//        return new FanoutExchange("ya.fanout")
        return build;
    }

this.rabbitTemplate.convertAndSend("simple.direct", "hi", "hello", new MessagePostProcessor() {
          @Override
          public Message postProcessMessage(Message message) throws AmqpException {
              //设置延迟消息
              message.getMessageProperties().setDelay(5000);
              return message;
          }
      });