SpringBoot整合RabbitMQ 手动应答(简单demo)

版本说明

• jdk 1.8
• rabbitmq 3.7.15 erlang 22.0
• springboot 2.3.3.release
• // todo 2021年1月8日 整理centos安装rabbitmq流程

1. 在rabbitmq的web管理界面，创建test队列

参数的含义

durability：是否持久化（重启或宕机后消息依然保存）

• durable 持久
• transient 暂时

新建maven项目。

2. pom.xml

<?xml version="1.0" encoding="utf-8"?>
<project xmlns="http://maven.apache.org/pom/4.0.0"
     xmlns:xsi="http://www.w3.org/2001/xmlschema-instance"
     xsi:schemalocation="http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelversion>4.0.0</modelversion>

  <parent>
    <groupid>org.springframework.boot</groupid>
    <artifactid>spring-boot-starter-parent</artifactid>
    <version>2.3.3.release</version>
    <relativepath/> <!-- lookup parent from repository -->
  </parent>

  <groupid>com.demo</groupid>
  <artifactid>rabbitmq-demo</artifactid>
  <version>1.0.0</version>

  <properties>
    <lombok.version>1.18.12</lombok.version>
  </properties>

  <dependencies>
    <!--web 模块-->
    <dependency>
      <groupid>org.springframework.boot</groupid>
      <artifactid>spring-boot-starter-web</artifactid>
    </dependency>

		<!-- amqp -->
    <dependency>
      <groupid>org.springframework.boot</groupid>
      <artifactid>spring-boot-starter-amqp</artifactid>
    </dependency>

    <!-- lombok-->
    <dependency>
      <groupid>org.projectlombok</groupid>
      <artifactid>lombok</artifactid>
      <optional>true</optional>
      <scope>provided</scope>
      <version>${lombok.version}</version>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-maven-plugin</artifactid>
      </plugin>
    </plugins>
  </build>
</project>

3. application.yaml

server:
 port: 20002
spring:
 rabbitmq:
 	# 这里我改了本地的hosts，实际地址是192.168.0.121
  host: vm.com
  port: 5672
  virtual-host: /
  username: admin
  password: admin
  # 开启消息确认模式
  # 消息发送到交换机确认机制，是否确认回调
  # publisher-confirms: true
  # 是否返回回调
  publisher-returns: true
  template:
   #开启mandatory: true, basic.return方法将消息返还给生产者
   mandatory: true
  listener:
   simple:
   	# 手动应答
    acknowledge-mode: manual
    # 最少消费者数量
    concurrency: 1 
    # 最多消费者数量
    max-concurrency: 10
    # 支持重试
    retry:
     enabled: true 

端口

• 5672：rabbitmq的通信端口
• 15672：web管理界面端口

4. rabbitmqdemo.java

@springbootapplication
@enablerabbit
public class rabbitmqdemoapplication {
  
  public static void main(string[] args) {
    springapplication.run(rabbitmqdemoapplication.class, args);
  }
}

5. rabbitconfig.java

@configuration
@slf4j
public class rabbitconfig {

  private rabbittemplate rabbittemplate;

  @bean
  public rabbittemplate rabbittemplate(connectionfactory connectionfactory) {
    rabbittemplate = new rabbittemplate(connectionfactory);
    return rabbittemplate;
  }

}

配置rabbitmq的消息模板。

6. 消息生产者 produce.java

@component
public class producer {
  
  // @qualifier("rabbittemplate")
  @autowired
  private rabbittemplate rabbittemplate;

  public void send() {
    for (int i = 0; i < 5; i++) {
      system.out.println("生产者发送消息，序号为: " + i);
      rabbittemplate.convertandsend("test", string.valueof(i));
    }
  }
}

初始化消息发送模板rabbittemplate，@qualifier注解用于限定具体的实现类，这里可以不指定。

7. 消息消费者 consumer.java

消费者1和消费者2均监听test队列。

不同的是，消费者1收到消息后返回确认应答basicack。

而消费者2收到消息后返回拒绝应答basicregect，消息被消费者拒绝后重新回到test队列中，等待下次发送给消费者。

@component
@slf4j
public class consumer {

  /**
   * 消费者1 模拟正常处理消息的情况，消息处理完毕发送确认应答
   * @param message
   * @param channel
   * @throws ioexception
   */
  @rabbitlistener(queues = "test")
  public void process1(message message, channel channel) throws ioexception {
    log.info("消费者1 接收消息: " + new string(message.getbody()));
    log.info("消费者1 确认应答消息：" + new string(message.getbody()));
    channel.basicack(message.getmessageproperties().getdeliverytag(), true);
  }

  /**
   * 消费者2 模拟处理消息出错的情况，消费者2向rabbitmq发送拒绝应答。
   * 处理失败的消息会被重新放入ready中，再次发送给消费者，直至收到确认应答
   * @param message
   * @param channel
   * @throws ioexception
   */
  @rabbitlistener(queues = "test")
  public void process2(message message, channel channel) throws ioexception {
    log.info("消费者2 接收消息：" + new string(message.getbody()));
    log.info("消费者2 拒绝应答消息：" + new string(message.getbody()));
    channel.basicreject(message.getmessageproperties().getdeliverytag(), true);
  }
}

8. 测试rabbitmqcontroller.java

@restcontroller
@requestmapping("")
public class rabbitmqcontroller {

  @autowired
  private producer producer;

  @getmapping("/send")
  public string send() {
    producer.send();
    return "发送完成";
  }
}

9. 测试

使用postman或浏览器使用get方法请求http://localhost:20001/send，生产者会向rabbitmq的test队列发送5条消息：

生产者发送消息，序号为: 0
生产者发送消息，序号为: 1
生产者发送消息，序号为: 2
生产者发送消息，序号为: 3
生产者发送消息，序号为: 4

可以看出序号为2的消息3次被消费者2接收，消费者2也3次发送拒绝应答，直到第4次才被消费者1接收，并返回确认应答。

到此这篇关于springboot整合rabbitmq 手动应答 简单demo的文章就介绍到这了,更多相关springboot整合rabbitmq 内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！