1、依賴 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> 2、rabbitmq鏈接配置 spring: r ...

1、依賴

	<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>

2、rabbitmq鏈接配置

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: wq
    password: qifeng
    virtual-host: /
    #開啟ack
    listener:
      direct:
        acknowledge-mode: manual
        prefetch: 1 # 限制一次拉取消息的數量
      simple:
        acknowledge-mode: manual #採取手動應答
        #concurrency: 1 # 指定最小的消費者數量
        #max-concurrency: 1 #指定最大的消費者數量
        retry:
          enabled: true # 是否支持重試

3、生產者

聲明交換機、隊列、綁定關於

package com.wanqi.mq;

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

/**
 * @Description TODO
 * @Version 1.0.0
 * @Date 2022/11/19
 * @Author wandaren
 */
@Configuration
public class RabbitMQConfig {
    public static final String EXCHANGE_NAME = "boot_topic_exchange";
    public static final String QUEUE_NAME = "boot_queue";
    //交換機
    @Bean("bootExchange")
    public Exchange bootExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true)
                .build();
    }
    //
    @Bean("bootQueue")
    public Queue bootQueue(){
        return QueueBuilder.durable(QUEUE_NAME).build();
    }
    //隊列和交換機綁定關係
    /*
        知道哪個隊列
        知道哪個交換機
        知道routing key
     */
    @Bean
    public Binding bindingQueueExchange(@Qualifier("bootQueue") Queue queue,
                                        @Qualifier("bootExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
    }
}

發送消息

package com.wanqi;

import com.wanqi.mq.RabbitMQConfig;
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;

@SpringBootTest
class SpringbootMqProducersApplicationTests {
	@Autowired
	private RabbitTemplate rabbitTemplate;

	@Test
	void contextLoads() {
		rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, "boot.test", "boot mq hello~~~");
	}

}

4、消費者

package com.wanqi.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @Description TODO
 * @Version 1.0.0
 * @Date 2022/11/19
 * @Author wandaren
 */
@Component
public class RabbitMQListener {

    @RabbitListener(queues = {"boot_queue"})
    public void listenerQueue(Object msg, Message message, Channel channel) {
        final long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            System.out.println(msg.toString());
            System.out.println(new String(message.getBody()));
//            int x = 3/0;
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            try {
                channel.basicNack(deliveryTag, true, true);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
        }
    }
}

5、死信隊列

消息成為死信的三種情況
1、隊列消息長度到達限制；
2、消費者拒接消費消息，並且不重回隊列；
3、原隊列存在消息過期設置，消息到達超時時間未被消費；

5.1、聲明交換機、隊列、綁定隊列

topic模式

package com.wanqi.mq;

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

/**
 * @Description TODO
 * @Version 1.0.0
 * @Date 2022/11/19
 * @Author wandaren
 */
@Configuration
public class QDRabbitMQConfig {
    //交換機名稱
    public static final String ITEM_EXCHANGE = "item_exchange";
    public static final String DEAD_EXCHANGE = "dead_exchange";
    //隊列名稱
    public static final String ITEM_QUEUE = "item_queue";
    public static final String DEAD_QUEUE = "dead_queue";

    //聲明業務交換機
    @Bean("itemExchange")
    public Exchange itemExchange(){
        return ExchangeBuilder.topicExchange(ITEM_EXCHANGE).durable(true).build();
    }
    //聲明死信交換機
    @Bean("deadExchange")
    public Exchange deadExchange(){
        return ExchangeBuilder.topicExchange(DEAD_EXCHANGE).durable(true).build();
    }

    /**
     * 聲明普通隊列，設置隊列消息過期時間，隊列長度，綁定的死信隊列
     */
    @Bean("itemQueue")
    public Queue itemQueue(){
        return QueueBuilder
                .durable(ITEM_QUEUE)
                // 隊列消息過期時間
                .ttl(10000)
                // 隊列長度
                .maxLength(15)
                // 聲明當前隊列綁定的死信交換機
                .deadLetterExchange(DEAD_EXCHANGE)
                // 聲明當前隊列死信轉發的路由key
                .deadLetterRoutingKey("infoDead.haha")
                .build();
    }
    /**
     * 死信隊列，消費者需要監聽的隊列
     */
    @Bean("deadQueue")
    public Queue deadQueue(){
        return QueueBuilder
                .durable(DEAD_QUEUE)
                .build();
    }

    //綁定隊列和交換機（業務）
    @Bean
    public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue,
                                     @Qualifier("itemExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("infoRouting.#").noargs();
    }
    //綁定隊列和交換機（死信）
    @Bean
    public Binding deadQueueExchange(@Qualifier("deadQueue") Queue queue,
                                     @Qualifier("deadExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("infoDead.#").noargs();
    }

}

5.2、發送測試消息

package com.wanqi;

import com.wanqi.mq.RabbitMQConfig;
import com.wanqi.mq.RabbitMQConfig2;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class SpringbootMqProducersApplicationTests {
	@Autowired
	private RabbitTemplate rabbitTemplate;
    
	// 原隊列存在消息過期設置，消息到達超時時間未被消費；
	@Test
	void contextLoads3() {
		MessageProperties messageProperties = new MessageProperties();
		// 設置過期時間，單位：毫秒
		messageProperties.setExpiration("5000");
		byte[] msgBytes = "rabbitmq ttl message ...".getBytes();
		Message message = new Message(msgBytes, messageProperties);

		//發送消息
		rabbitTemplate.convertAndSend(QDRabbitMQConfig.ITEM_EXCHANGE,"infoRouting.hehe",message);
		System.out.println("發送消息成功");
	}
    // 模擬隊列消息長度到達限制
	@Test
	void contextLoads4() {
		for (int i = 0; i < 20; i++) {
			rabbitTemplate.convertAndSend(QDRabbitMQConfig.ITEM_EXCHANGE,"infoRouting.hehe",i + "---message");

		}
	}
}

5.3、消費者

package com.wanqi.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @Description TODO
 * @Version 1.0.0
 * @Date 2022/11/19
 * @Author wandaren
 */
@Component
public class RabbitMQListener {
    @RabbitListener(queues = {"dead_queue"})
    public void listenerQueue2(Object msg, Message message, Channel channel) {
        final long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            System.out.println(msg.toString());
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            try {
                channel.basicNack(deliveryTag, true, true);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
        }
    }
}

6、延遲隊列（TTL + 死信隊列）

6.1、聲明交換機、隊列、綁定隊列

direct模式


package com.wanqi.mq;

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

/**
 * @Description TODO
 * @Version 1.0.0
 * @Date 2022/11/19
 * @Author wandaren
 */
@Configuration
public class TtlQueueConfig {
    //普通交換機名稱
    public static final String X_CHANGE = "X_Exchange";
    //死信交換機名稱
    public static final String Y_DEAD_CHANGE = "Y_Exchange";
    //普通隊列
    public static final String QUEUE_A = "QA_QUEUE";
    public static final String QUEUE_B = "QB_QUEUE";
    //死信隊列
    public static final String DEAD_QUEUE_D = "QD_QUEUE";

    //聲明普通交換機
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return ExchangeBuilder.directExchange(X_CHANGE).durable(true)
                .build();
    }

    //聲明死信交換機
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return ExchangeBuilder.directExchange(Y_DEAD_CHANGE).durable(true)
                .build();
    }

    /**
     * 聲明隊列,延遲10秒
     */

    @Bean("queueA")
    public Queue queueA() {

        return QueueBuilder.durable(QUEUE_A)
                .deadLetterExchange(Y_DEAD_CHANGE) //死信交換機
                .deadLetterRoutingKey("YD")  //死信RoutingKey
                .ttl(10000)  //消息過期時間
                .build();
    }

    /**
     * 聲明隊列，延遲40秒
     *
     * @return
     */
    @Bean("queueB")
    public Queue queueB() {
        return QueueBuilder.durable(QUEUE_B)
                .deadLetterExchange(Y_DEAD_CHANGE) //死信交換機
                .deadLetterRoutingKey("YD")  //死信RoutingKey
                .ttl(40000)  //消息過期時間
                .build();
    }


    /**
     * 死信隊列，消費者需要監聽的隊列
     */
    @Bean("queueD")
    public Queue queueD() {
        return QueueBuilder.durable(DEAD_QUEUE_D).build();
    }


    //綁定  X_CHANGE綁定queueA
    @Bean
    public Binding queueABindingX(@Qualifier("queueA") Queue queueA, @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }

    //綁定  X_CHANGE綁定queueB
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB, @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    }


    //綁定  Y_CHANGE綁定queueD
    @Bean
    public Binding queueDBindingY(@Qualifier("queueD") Queue queueD, @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }
}

6.2、發送消息

    @Test
    void contextLoads5() {
        String message = "延遲隊列消息";
        rabbitTemplate.convertAndSend(TtlQueueConfig.X_CHANGE, "XA", "TTL=10s的隊列：" + message);
        rabbitTemplate.convertAndSend(TtlQueueConfig.X_CHANGE, "XB", "TTL=40s的隊列：" + message);
    }

6.3、消費

package com.wanqi.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

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

/**
 * @Description TODO
 * @Version 1.0.0
 * @Date 2022/11/19
 * @Author wandaren
 */
@Component
public class RabbitMQListener {
    @RabbitListener(queues = "QD_QUEUE")
    public void receiveMessage(Object msg,Message message, Channel channel){
        final long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            System.out.println(msg.toString());
            System.out.print("當前時間: " + new Date());
            System.out.println("  收到死信隊列的消息：" + msg);
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            try {
                channel.basicNack(deliveryTag, true, true);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
        }
    }
}

Springboot整合RabbitMQ基本使用

1、依賴

2、rabbitmq鏈接配置

3、生產者

4、消費者

5、死信隊列

5.1、聲明交換機、隊列、綁定隊列

5.2、發送測試消息

5.3、消費者

6、延遲隊列（TTL + 死信隊列）

6.1、聲明交換機、隊列、綁定隊列

6.2、發送消息

6.3、消費