版本说明

• 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

  4.0.0

  
    org.springframework.boot
    spring-boot-starter-parent
    2.3.3.RELEASE
     
  

  com.demo
  rabbitmq-demo
  1.0.0

  
    1.18.12
  

  
    
    
      org.springframework.boot
      spring-boot-starter-web
    

		
    
      org.springframework.boot
      spring-boot-starter-amqp
    

    
    
      org.projectlombok
      lombok
      true
      provided
      ${lombok.version}
    
  

  
    
      
        org.springframework.boot
        spring-boot-maven-plugin
      
    
  

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接收，并返回确认应答。