SpringBoot整合RabbitMQ

2023-11-17 13:45:34

导入依赖

 <!--springboot整合rabbitmq的依赖-->
 <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>

一、简单整合

我们这里在两个项目中完成一个生产者项目，一个消费者项目

生产者

配置文件

server.port=8080
spring.rabbitmq.host=192.168.195.157
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

controller（发送端）

/**
 * 生产者模拟请求发送
 * @author acoffee
 * @create 2022-01-12 12:00
 */
@RestController
public class SendController {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @RequestMapping("/sendmail")
    public String sendmail(){
        //发送消息
        rabbitTemplate.convertAndSend("exchangetopic","email","email");
        return "sendmail";
    }

    @RequestMapping("/sendsms")
    public String sendSms(){
        //发送消息
        rabbitTemplate.convertAndSend("exchangetopic","sms","sms");
        return "sendsms";
    }

    @RequestMapping("/sendall")
    public String sendAll(){
        //发送消息
        rabbitTemplate.convertAndSend("exchangetopic","sms.email","sms.email");
        return "sendall";
    }
}

消费者

配置文件

server.port=8081
spring.rabbitmq.host=192.168.195.157
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

配置类

/**
 * Rabbitmq配置类,定义交换机,队列,绑定
 *
 * @author acoffee
 * @create 2022-01-12 12:12
 */
@Configuration
public class RabbitmqConfiguration {

    //定义交换机
    @Bean
    public TopicExchange exchangeTopic() {
        return new TopicExchange("exchangetopic");
    }

    //定义队列
    @Bean
    public Queue queueMail() {
        return new Queue("queueemail");
    }

    @Bean
    public Queue queueMsg() {
        return new Queue("queuesms");
    }

    @Bean
    public Binding bindingEmail(TopicExchange exchange, Queue queueMail) {
        return BindingBuilder.bind(queueMail).to(exchange).with("#.email.#");
    }

    @Bean
    public Binding bindingMsg(TopicExchange exchange, Queue queueMsg) {
        return BindingBuilder.bind(queueMsg).to(exchange).with("#.sms.#");
    }
}

消费者处理请求

/**
 * 定义一个bean,消费者,监听队列,定义处理消息的方法
 *
 * @author acoffee
 * @create 2022-01-12 13:05
 */
@Component
public class MyConsumer {

    //指定当前方法监听那个队列
    //String msg: 消息体
    //Message: 消息对象
    //Channel: 信道
    @RabbitListener(queues = "queueemail")
    public void receiveEmail(String msg, Message message, Channel channel) {
        System.out.println("处理Email");
        System.out.println("msg:" + msg);
        System.out.println("message:" + message);
        System.out.println("channel:" + channel);
        System.out.println("---------------");
    }

    @RabbitListener(queues = "queuesms")
    public void receiveSms(String msg, Message message, Channel channel) {
        System.out.println("处理Sms");
        System.out.println("msg:" + msg);
        System.out.println("message:" + message);
        System.out.println("channel:" + channel);
        System.out.println("---------------");
    }
}

执行结果：
SpringBoot整合RabbitMQ
SpringBoot整合RabbitMQ

二、生产者消息确认

配置文件

server.port=8080
spring.rabbitmq.host=192.168.195.157
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
#生产者confrim
#spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-confirm-type=correlated
#生产者return
spring.rabbitmq.publisher-returns=true

添加配置类开启监听

package com.acoffee.configuration;

import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * 实现配置类型，实现接口方法，启动生产者确认
 * @author acoffee
 * @create 2022-01-12 15:09
 */
@Configuration
public class CallbackConfiguration implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    @Autowired
    RabbitTemplate rabbitTemplate;

    //@PostConstruct @PreDestroy：类似生命周期函数
    @PostConstruct
    public void setUp(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("confirm");
        System.out.println("correlationData:"+correlationData);
        System.out.println("ack:"+ack);
        System.out.println("cause:"+cause);
        System.out.println("-----------------");
    }

    @Override
    public void returnedMessage(ReturnedMessage returned) {
        System.out.println("Return");
        System.out.println("message:"+returned.getMessage());
        System.out.println("replyText:"+returned.getReplyText());
        System.out.println("exchange:"+returned.getExchange());
        System.out.println("routingKey"+returned.getRoutingKey());
        System.out.println("-----------------");
    }
}

执行结果：

我们测试了三种

第一种：故意将exchange的名字写错，可以看出返回的 ack 为 false 就是我们说的 nack，并且从cause中就可以看出错误原因，我们原来也说过在基础api中是看不到nack的，在springboot中是可以的。
SpringBoot整合RabbitMQ
第二种：故意写错routingKey，从结果可以看出的ack为true，说明的已经到交换机了，但是由于通过routingKey找不到对应的管道所以给我们返回了消息，在replyText中也可以看到replyText:NO_ROUTE造成的
SpringBoot整合RabbitMQ第三种：正确情况
只返回了confirm中的信息，可以看出ack为true，并且消费者也接收并处理了消息

SpringBoot整合RabbitMQ
SpringBoot整合RabbitMQ

三、消费者消息确认

自动确认

默认就是自动确认，一下全取出来，宕机就会出现问题

SpringBoot整合RabbitMQ

手动确认+限流

消费者配置文件

server.port=8081
spring.rabbitmq.host=192.168.195.157
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
#消费者手动确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
#限流
spring.rabbitmq.listener.simple.prefetch=1

@RabbitListener(queues = "queueemail")
public void receiveEmail(String msg, Message message, Channel channel) throws Exception {

    System.out.println("处理Email");
    System.out.println("msg:" + msg);
    System.out.println("message:" + message);
    System.out.println("channel:" + channel);
    //手动确认
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    System.out.println("---------------");
}

执行结果：
一个一个读取，一个一个确认
SpringBoot整合RabbitMQ

四、消息属性

@RequestMapping("/sendmail")
public String sendmail() {
    	//发送消息
        rabbitTemplate.convertAndSend("exchangetopic", "email", "email1",
                //设置消息属性
                new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        message.getMessageProperties().setExpiration("10000");
                        //message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                        message.getMessageProperties().getHeaders().put("h1", "v1");
                        message.getMessageProperties().getHeaders().put("h2", "v2");
                        return message;
                    }
                });
    return "sendmail";
}

执行结果：
SpringBoot整合RabbitMQ

五、对象存储

配置类中添加JSON消息转化器，底层默认使用的Jackson

 @Bean
 public MessageConverter messageConverter(){
     return new Jackson2JsonMessageConverter();
 }

消费者和生产者都要配置实体类，且包名和类名必须一样

生产者

@RequestMapping("/sendsms")
public String sendSms() {
    //发送消息:传对象
    rabbitTemplate.convertAndSend("exchangetopic", "sms", new Student(1,"tom"));
    return "sendsms";
}

消费者

@RabbitListener(queues = "queuesms")
public void receiveSms(Student msg, Message message, Channel channel) throws IOException {
    System.out.println("处理Sms");
    System.out.println("msg:" + msg);
    System.out.println("message:" + message);
    System.out.println("channel:" + channel);
    //手动确认
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    System.out.println("---------------");
}

执行结果：
SpringBoot整合RabbitMQ

六、完整代码

依赖

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

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!--springboot整合rabbitmq的依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

生产者

server.port=8080
spring.rabbitmq.host=192.168.195.157
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
#生产者confrim
#spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-confirm-type=correlated
#生产者return
spring.rabbitmq.publisher-returns=true
#解决循环依赖问题2.6.x可能会出现
spring.main.allow-circular-references=true

实体类

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Student implements Serializable {
    int id;
    String name;
}

配置类

@Configuration
public class CallbackConfiguration implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    @Autowired
    RabbitTemplate rabbitTemplate;

    //@PostConstruct @PreDestroy：类似生命周期函数
    @PostConstruct
    public void setUp(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("confirm");
        System.out.println("correlationData:"+correlationData);
        System.out.println("ack:"+ack);
        System.out.println("cause:"+cause);
        System.out.println("-----------------");
    }

    @Override
    public void returnedMessage(ReturnedMessage returned) {
        System.out.println("Return");
        System.out.println("message:"+returned.getMessage());
        System.out.println("replyText:"+returned.getReplyText());
        System.out.println("exchange:"+returned.getExchange());
        System.out.println("routingKey"+returned.getRoutingKey());
        System.out.println("-----------------");
    }

    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

controller层(发送请求)

import com.acoffee.entity.Student;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 生产者模拟请求发送
 *
 * @author acoffee
 * @create 2022-01-12 12:00
 */
@RestController
public class SendController {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @RequestMapping("/sendmail")
    public String sendmail() {
        //发送消息
        for (int i = 1; i <= 10; i++) {
            rabbitTemplate.convertAndSend("exchangetopic", "email", "email" + i,
                    //设置消息属性
                    new MessagePostProcessor() {
                        @Override
                        public Message postProcessMessage(Message message) throws AmqpException {
                            //过期时间
                            message.getMessageProperties().setExpiration("10000");
                            //message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                            message.getMessageProperties().getHeaders().put("h1", "v1");
                            message.getMessageProperties().getHeaders().put("h2", "v2");
                            return message;
                        }
                    });
        }
        return "sendmail";
    }

    @RequestMapping("/sendsms")
    public String sendSms() {
        //发送消息:传对象
        rabbitTemplate.convertAndSend("exchangetopic", "sms", new Student(1,"tom"));
        return "sendsms";
    }

    @RequestMapping("/sendall")
    public String sendAll() {
        //发送消息
        rabbitTemplate.convertAndSend("exchangetopic", "sms.email", "sms.email");
        return "sendall";
    }
}

消费者

实体类

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Student implements Serializable {
    int id;
    String name;
}

配置类

/**
 * Rabbitmq配置类,定义交换机,队列,绑定
 *
 * @author acoffee
 * @create 2022-01-12 12:12
 */
@Configuration
public class RabbitmqConfiguration {

    //定义交换机
    @Bean
    public TopicExchange exchangeTopic() {
        return new TopicExchange("exchangetopic");
    }

    //定义队列
    @Bean
    public Queue queueMail() {
        return new Queue("queueemail");
    }

    @Bean
    public Queue queueMsg() {
        return new Queue("queuesms");
    }

    @Bean
    public Binding bindingEmail(TopicExchange exchange, Queue queueMail) {
        return BindingBuilder.bind(queueMail).to(exchange).with("#.email.#");
    }

    @Bean
    public Binding bindingMsg(TopicExchange exchange, Queue queueMsg) {
        return BindingBuilder.bind(queueMsg).to(exchange).with("#.sms.#");
    }

    //对象转json
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

MyConsumer 类

/**
 * 定义一个bean,消费者,监听队列,定义处理消息的方法
 *
 * @author acoffee
 * @create 2022-01-12 13:05
 */
@Component
public class MyConsumer {
    
    //指定当前方法监听那个队列
    //String msg: 消息体
    //Message: 消息对象
    //Channel: 信道
    @RabbitListener(queues = "queueemail")
    public void receiveEmail(String msg, Message message, Channel channel) throws Exception {

        Thread.sleep(10000);

        System.out.println("处理Email");
        System.out.println("msg:" + msg);
        System.out.println("message:" + message);
        System.out.println("channel:" + channel);
        //手动确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        System.out.println("---------------");
    }
    
    @RabbitListener(queues = "queuesms")
    public void receiveSms(Student msg, Message message, Channel channel) throws IOException {
        System.out.println("处理Sms");
        System.out.println("msg:" + msg);
        System.out.println("message:" + message);
        System.out.println("channel:" + channel);
        //手动确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        System.out.println("---------------");
    }
}
上一篇：RabbitMQ单机多实例集群搭建


下一篇：RabbitMQ简单安装