RabbitMQ快速使用

安裝#

	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 Boot 集成#

example:springboot-middlewave-example/springboot-rabbitmq at master · liyown/springboot-middlewave-example (github.com)

發送者#

發送者超時重連（異步）配置

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

使用配置:

配置消息轉化服務，默認是 JDK 序列化，不易讀，占用空間大。兼容性不高

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

設置消息發送回調
回調有兩種情況，一是到了交換機就會返回 confirm 信息，如果沒有到達隊列，將會回調 ReturnsCallback，此時一般是運維層面的问题

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

JAVA API 使用

帶回調的使用：

public void testRabiitStart() throws InterruptedException {  
  
    CorrelationData  correlationData = new CorrelationData("1");  
    // 設置回調  
    rabbitTemplate.setConfirmCallback((correlationData1, ack, cause) -> {  
        System.out.println("cause: " + cause);  
        System.out.println("correlationData: " + correlationData1);  
        if (ack) {  
            System.out.println("消息發送成功");  
        } else {  
            System.out.println("消息發送失敗");  
        }  
    });  
    // 隊列名  
    String queueName = "queue.lyw";  
  
  
    // 發送消息, 如果沒有隊列會自動創建  
    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 配置

spring:  
  application:  
    name: publisher  
  rabbitmq:  
    host: 192.168.208.128  
    port: 5672  
    username: user  
    password: password  
    virtual-host: /hmall  
    listener:  
      simple:  
        # 表示自動回應，當發送異常，將重試，重試不成功發送到死信隊列（三種策略）
        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("工作隊列2接收到消息: {}，時間：{},線程{}", message, LocalDateTime.now(), thread.getName());  
	}

死信隊列的初始化
當到達重試次數後，有三種策略，分別為
image.png|600

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

延遲隊列的使用#

安裝插件（github）
rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ (github.com)
複製插件到插件文件夾
安裝插件

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;  
	    });  
	}

消費延遲消息：和普通隊列一致

此文由 Mix Space 同步更新至 xLog
原始鏈接為 https://me.liuyaowen.club/posts/default/20240820and1