Springboot 整合RabbitMq(用心看完这一篇就够了)

RabbitMQ 是一个开源的消息队列系统，支持多种消息递协议。在 RabbitMQ 中，多种消息模型可以用于不同的场。本文将详细讲解 SpringBoot 整合 RabbitMQ 的完整攻略，并提供两个示例说明。

环境准备

在开始之前，需要确保已经安装了以下环境：

• JDK 1.8 或以上版本
• Maven 3.0 或以上版本
• RabbitMQ 服务器

示例一：使用 SpringBoot 实现工作队列

在本例中，我们将使用 SpringBoot 实现工作队列。具体步骤如下：

1. 创建一个 SpringBoot 项目添加 RabbitMQ 依赖。
2. 创建一个 RabbitMQ 的生产者并将消息设置为持久化消息。
3. 创建一个 RabbitMQ 的消费者并确认消息已被接收。
4. 创建一个队列并将其绑定到一个交换机上。
5. 发送消息到队列中。

1. 创建一个 SpringBoot 项目并添加 RabbitMQ 依赖

在 pom.xml 文件中添加以下依赖：

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

2. 创建一个 RabbitMQ 的生产者并将消息设置为持久化消息

@Component
public class RabbitMQProducer {

    private final RabbitTemplate rabbitTemplate;

    public RabbitMQProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void send(String message) {
        rabbitTemplate.convertAndSend("myQueue", message, message1 -> {
            MessageProperties messageProperties = message1.getMessageProperties();
            messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message1;
        });
    }
}

在上述代码中，我们创建了一个 RabbitMQ 的生产者并将消息设置为持久化消息。在 send 方法中，我们使用 rabbitTemplate 发送消息到队列中，并将消息设置为持久化消息。

3. 创建一个 RabbitMQ 的消费者并确认消息已被接收

@Component
public class RabbitMQConsumer {

    @RabbitListener(queues = "myQueue")
    public void receive(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        System.out.println("Received message: " + message);
        try {
            doWork(message);
            channel.basicAck(tag, false);
        } catch (Exception e) {
            channel.basicNack(tag, false, true);
        }
    }

    private void doWork(String message) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

在上述代码中，我们创建了一个 RabbitMQ 的消费者并确认消息已被接收。在 receive 方法中，我们处理消息并确认消息已被消费。如果任务处理失败，我们将使用 channel.basicNack 方法将消息重新放回队列中。

4. 创建一个队列并将其绑定到一个交换机上

@Configuration
public class RabbitMQConfig {

    @Bean
    public Queue myQueue() {
        return new Queue("myQueue", true);
    }

    @Bean
    public DirectExchange myExchange() {
        return new DirectExchange("myExchange");
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(myQueue()).to(myExchange()).with("myRoutingKey");
    }
}

在上述代码中，我们创建了一个队列并将其绑定到一个交换机上。在 myQueue 方法中，我们创建了一个队列并将其设置为持久化队列。在 myExchange 方法中，我们创建了一个直连交换机。在 binding 方法中，我们将队列绑定到交换机上，并指定了一个路由键。

5. 发送消息到队列中

@SpringBootApplication
public class Application implements CommandLineRunner {

    private final RabbitMQProducer rabbitMQProducer;

    public Application(RabbitMQProducer rabbitMQProducer) {
        this.rabbitMQProducer = rabbitMQProducer;
    }

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        for (int i = 0; i < 10; i++) {
            String message = "Message " + i;
            rabbitMQProducer.send(message);
            System.out.println("Sent message: " + message);
        }
    }
}

在上述代码中，我们创建了一个 SpringBoot 应用程序，并在 run 方法中发送了 10 条消息到队列中。

示例二：使用 SpringBoot 实现发布/订阅模式

在本例中，我们将使用 SpringBoot 实现发布/订阅模式。具体步骤如下：

1. 创建一个 SpringBoot 项目并添加 RabbitMQ 依赖。
2. 创建一个 RabbitMQ 的生产者并将消息设置为持久化消息。
3. 创建一个 RabbitMQ 的消费者并确认消息已被接收。
4. 创建一个交换机并将其绑定到多个队列上。
5. 发送消息到交换机中。

1. 创建一个 SpringBoot 项目并添加 RabbitMQ 依赖

在 pom.xml 文件中添加以下依赖：

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

2. 创建一个 RabbitMQ 的生产者并将消息设置为持久化消息

@Component
public class RabbitMQProducer {

    private final RabbitTemplate rabbitTemplate;

    public RabbitMQProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void send(String message) {
        rabbitTemplate.convertAndSend("myExchange", "myRoutingKey", message, message1 -> {
            MessageProperties messageProperties = message1.getMessageProperties();
            messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message1;
        });
    }
}

在上述代码中，我们创建了一个 RabbitMQ 的生产者并将消息设置为持久化消息。在 send 方法中，我们使用 rabbitTemplate 发送消息到交换机中，并将消息设置为持久化消息。

3. 创建一个 RabbitMQ 的消费者并确认消息已被接收

@Component
public class RabbitMQConsumer {

    @RabbitListener(queues = "myQueue1")
    public void receive1(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        System.out.println("Received message from queue1: " + message);
        try {
            doWork(message);
            channel.basicAck(tag, false);
        } catch (Exception e) {
            channel.basicNack(tag, false, true);
        }
    }

    @RabbitListener(queues = "myQueue2")
    public void receive2(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        System.out.println("Received message from queue2: " + message);
        try {
            doWork(message);
            channel.basicAck(tag, false);
        } catch (Exception e) {
            channel.basicNack(tag, false, true);
        }
    }

    private void doWork(String message) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

在上述代码中，我们创建了两个 RabbitMQ 的消费者并确认消息已被接收。在 receive1 方法中，我们处理来自队列1的消息并确认消息已被消费。在 receive2 方法中，我们处理来自队列2的消息并确认消息已被消费。如果任务处理失败，我们将使用 channel.basicNack 方法将消息重新放回队列中。

4. 创建一个交换机并将其绑定到多个队列上

@Configuration
public class RabbitMQConfig {

    @Bean
    public Queue myQueue1() {
        return new Queue("myQueue1", true);
    }

    @Bean
    public Queue myQueue2() {
        return new Queue("myQueue2", true);
    }

    @Bean
    public FanoutExchange myExchange() {
        return new FanoutExchange("myExchange");
    }

    @Bean
    public Binding binding1() {
        return BindingBuilder.bind(myQueue1()).to(myExchange());
    }

    @Bean
    public Binding binding2() {
        return BindingBuilder.bind(myQueue2()).to(myExchange());
    }
}

在上述代码中，我们创建了一个交换机并将其绑定到多个队列上。在 myQueue1 方法中，我们创建了一个队列并将其设置为持久化队列。在 myQueue2 方法中，我们创建了另一个队列并将其设置为持久化队列。在 myExchange 方法中，我们创建了一个扇形交换机。在 binding1 方法中，我们将队列1绑定到交换机上。在 binding2 方法中，我们将队列2绑定到交换机上。

5. 发送消息到交换机中

@SpringBootApplication
public class Application implements CommandLineRunner {

    private final RabbitMQProducer rabbitMQProducer;

    public Application(RabbitMQProducer rabbitMQProducer) {
        this.rabbitMQProducer = rabbitMQProducer;
    }

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        for (int i = 0; i < 10; i++) {
            String message = "Message " + i;
            rabbitMQProducer.send(message);
            System.out.println("Sent message: " + message);
        }
    }
}

在上述代码中，我们创建了一个 SpringBoot 应用程序，并在 run 方法中发送了 10 条消息到交换机中。这些消息将被广播到所有绑定到交换机上的队列中。