刘耀文

	docker run --name rabbitmq \
	-p 5672:5672 \
	-p 15672:15672 \
	--network hm-net \
	--hostname my-rabbit \
	-e RABBITMQ_DEFAULT_USER=user \
	-e RABBITMQ_DEFAULT_PASS=password \
	-v mq_plugins:/plugins \
	-d rabbitmq:3.8-management

spring:  
  application:  
    name: publisher  
  rabbitmq:  
    host: 192.168.208.128  
    port: 5672  
    username: user  
    password: password  
    virtual-host: /hmall  
    # Send confirm mechanism
    publisher-confirm-type: correlated  
    publisher-returns: true  
    # Retry mechanism
    template:  
      retry:  
        enabled: true  
        max-attempts: 3  
        initial-interval: 1000  
        max-interval: 10000  
        multiplier: 2  
server:  
  port: 7081

	@Bean  
	public MessageConverter messageConverter() {  
	    Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();  
	    jackson2JsonMessageConverter.setCreateMessageIds(true);  
	    return  jackson2JsonMessageConverter;  
	}

public void init() {  
    rabbitTemplate.setReturnsCallback(  
            (ReturnedMessage returned) -> {  
                System.out.println("Message lost: " + returned.getMessage());  
            }  
    );  
    rabbitTemplate.setBeforePublishPostProcessors((message) -> {  
        MessageProperties messageProperties = message.getMessageProperties();  
        String messageId = messageProperties.getMessageId();  
        System.out.println("messageId: " + messageId);  
        return message;  
    });  
  
}

public void testRabiitStart() throws InterruptedException {  
  
    CorrelationData  correlationData = new CorrelationData("1");  
    // Set callback  
    rabbitTemplate.setConfirmCallback((correlationData1, ack, cause) -> {  
        System.out.println("cause: " + cause);  
        System.out.println("correlationData: " + correlationData1);  
        if (ack) {  
            System.out.println("Message sent successfully");  
        } else {  
            System.out.println("Message sending failed");  
        }  
    });  
    // Queue name  
    String queueName = "queue.lyw";  
  
  
    // Send message, if there is no queue, it will be created automatically  
    rabbitTemplate.convertAndSend("direct.lyw", queueName, "hello rabbitmq", correlationData);  
  
}

public void testRabiitMqDelayMessage() {  
    CorrelationData  correlationData = new CorrelationData("1");  
    rabbitTemplate.convertAndSend("delay.direct", "delay.queue", "hello rabbitmq delay",  
                                  (Message message) -> {  
                                      message.getMessageProperties().setDelayLong(10000L);  
                                      return message;  
                                  }  
                                  , correlationData);  
}

spring:  
  application:  
    name: publisher  
  rabbitmq:  
    host: 192.168.208.128  
    port: 5672  
    username: user  
    password: password  
    virtual-host: /hmall  
    listener:  
      simple:  
        # Indicates automatic acknowledgment, when sending fails, it will retry, if retries fail, send to dead letter queue (three strategies)
        acknowledge-mode: auto  
        retry:  
          enabled: true  
          max-attempts: 3  
          initial-interval: 1000  
          max-interval: 10000  
          multiplier: 2  
          stateless: true  
server:  
  port: 7082

	@Bean  
	public DirectExchange directExchange() {  
	    return new DirectExchange("direct.lyw");  
	}  
	  
	@Bean  
	public Queue queue() {  
	    return new Queue("queue.lyw");  
	}  
	  
	@Bean  
	public Binding binding(Queue queue, FanoutExchange fanoutExchange) {  
	    return BindingBuilder.bind(queue).to(fanoutExchange);  
	}

	@RabbitListener(queues = "work.queue1")  
	public void onMessageWorkerQueue2(String message) throws InterruptedException {  
	    Thread.sleep(200);  
	    Thread thread = Thread.currentThread();  
	    log.info("Work queue 2 received message: {}，time：{}, thread{}", message, LocalDateTime.now(), thread.getName());  
	}

	@Bean  
	public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){  
	    return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");  
	}

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

	@Bean  
	public Queue delayQueue() {  
	    return new Queue("delay.queue");  
	}  
	  
	@Bean  
	public DirectExchange delayExchange() {  
	    return ExchangeBuilder.directExchange("delay.direct").delayed().durable(true).build();  
	}  
	  
	@Bean Binding delayBinding(Queue delayQueue, DirectExchange delayExchange) {  
	    return BindingBuilder.bind(delayQueue).to(delayExchange).with("delay.queue");  
	}

	@Test  
	public void testRabiitMqDelayMessage() {  
	    CorrelationData  correlationData = new CorrelationData("1");  
	    rabbitTemplate.convertAndSend("delay.direct", "delay.queue", "hello rabbitmq delay",  
	                                  (Message message) -> {  
	                                      message.getMessageProperties().setDelayLong(10000L);  
	                                      return message;  
	                                  }  
	                                  , correlationData);  
	}  
	  
	@Test  
	public void testRabiitMqDelayMessage2() {  
	    rabbitTemplate.convertAndSend("delay.direct", "delay.queue", "1812793267355439105", message -> {  
	        message.getMessageProperties().setDelayLong(10000L);  
	        return message;  
	    });  
	}