Rabbitmq消息丢失-消费者消息丢失（二）

说明：消费端在处理消息的过程中出现异常，例如：业务逻辑异常，或者消费者被停机，或者网络断开连接等，以上等情况使消息没有得到正确恰当的处理，也会使消息丢失。

分析：分析就是说明中的例如！

解决：ACK确认机制

所谓的ACK就是：首先关闭自动确认【自动ACK】，消费者收到一个消息后，就可以发一个确认【ACK】给MQ，当然什么时候发送确认【ACK】是程序员决定的，也就是说每次在确保处理完这个消息相关的业务后，程序员可以手动发送确认【ACK】，之后把消息从MQ中干掉！这样即使出现了异常也可以有效的消费消息。

工程图：

1.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>
    <artifactId>spring-boot-starter-parent</artifactId>  <!-- 被继承的父项目的构件标识符 -->
    <groupId>org.springframework.boot</groupId>  <!-- 被继承的父项目的全球唯一标识符 -->
    <version>2.2.2.RELEASE</version>  <!-- 被继承的父项目的版本 -->
  </parent>

  <groupId>MqLossDemoTwo</groupId>
  <artifactId>MqLossDemoTwo</artifactId>
  <version>1.0-SNAPSHOT</version>
  <packaging>war</packaging>

  <name>MqLossDemoTwo Maven Webapp</name>
  <!-- FIXME change it to the project's website -->
  <url>http://www.example.com</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
  </properties>

  <dependencies>
    <!--spring boot核心-->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <!--spring boot 测试-->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>
    <!--springmvc web-->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!--开发环境调试-->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-devtools</artifactId>
      <optional>true</optional>
    </dependency>
    <!--amqp 支持-->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

    <!--redis-->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-data-redis</artifactId>
      <version>2.1.7.RELEASE</version>
    </dependency>

    <dependency>
      <groupId>com.alibaba</groupId>
      <artifactId>fastjson</artifactId>
      <version>1.2.78</version>
    </dependency>

    <dependency>
      <groupId>commons-lang</groupId>
      <artifactId>commons-lang</artifactId>
      <version>2.6</version>
    </dependency>

    <!--lombok-->
    <dependency>
      <groupId>org.projectlombok</groupId>
      <artifactId>lombok</artifactId>
      <version>1.16.10</version>
    </dependency>
  </dependencies>


  <build>
    <finalName>MqLossDemoTwo</finalName>
    <pluginManagement><!-- lock down plugins versions to avoid using Maven defaults (may be moved to parent pom) -->
      <plugins>
        <plugin>
          <artifactId>maven-clean-plugin</artifactId>
          <version>3.1.0</version>
        </plugin>
        <!-- see http://maven.apache.org/ref/current/maven-core/default-bindings.html#Plugin_bindings_for_war_packaging -->
        <plugin>
          <artifactId>maven-resources-plugin</artifactId>
          <version>3.0.2</version>
        </plugin>
        <plugin>
          <artifactId>maven-compiler-plugin</artifactId>
          <version>3.8.0</version>
        </plugin>
        <plugin>
          <artifactId>maven-surefire-plugin</artifactId>
          <version>2.22.1</version>
        </plugin>
        <plugin>
          <artifactId>maven-war-plugin</artifactId>
          <version>3.2.2</version>
        </plugin>
        <plugin>
          <artifactId>maven-install-plugin</artifactId>
          <version>2.5.2</version>
        </plugin>
        <plugin>
          <artifactId>maven-deploy-plugin</artifactId>
          <version>2.8.2</version>
        </plugin>
      </plugins>
    </pluginManagement>
  </build>
</project>

2.application.yml

server:
  port: 8080
spring:
  rabbitmq:
    port: 5672
    host: 你的 rabbitmq IP
    username: admin
    password: admin
    virtual-host: /
    listener:
      simple:
        concurrency: 10
        max-concurrency: 10
        prefetch: 1
        auto-startup: true
        default-requeue-rejected: true
        # 设置消费端手动 ack
        acknowledge-mode: manual
        # 是否支持重试
        retry:
          enabled: true

3.RabbitMqQueueConfig

package com.dev.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 类名称：
 *
 * @author 李庆伟
 * @date 2024年03月04日 14:12
 */
@Configuration
public class RabbitMqQueueConfig {

    //绑定键
    public final static String QUEUE_ONE = "loss_queue";

    public final static String EXCHANGE_ONE = "loss_exchange";



    @Bean
    public Queue directQueue() {
        return new Queue(RabbitMqQueueConfig.QUEUE_ONE);
    }

    //Direct交换机 起名：directExchange
    @Bean
    DirectExchange directExchange() {
        return new DirectExchange(RabbitMqQueueConfig.EXCHANGE_ONE,true,false);
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：directRoutingKey
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(directQueue()).to(directExchange()).with("directRoutingKey");
    }


}

4.RabbitController

package com.dev.controller;

import com.alibaba.fastjson.JSONObject;
import com.dev.config.RabbitMqQueueConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;


/**
 * 类名称：消息丢失问题
 *
 * @author lqw
 * @date 2024年02月27日 14:47
 */
@Slf4j
@RestController
@RequestMapping("loss")
public class RabbitController {


    @Autowired
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法


    /**
     * 消息丢失
     * @return
     */
    @GetMapping("/sendMessage")
    public String sendMessage() {
        String id = UUID.randomUUID().toString().replace("-","");
        Map<String,Object> addMap = new HashMap<>();//添加用户信息
        addMap.put("id",id);
        addMap.put("name","张龙");
        Message msg = MessageBuilder.withBody(JSONObject.toJSONString(addMap).getBytes()).setMessageId(id).build();

        rabbitTemplate.convertAndSend(RabbitMqQueueConfig.EXCHANGE_ONE, "directRoutingKey", msg);
        return "ok";
    }





}

5.RabbitMqListener

package com.dev.listener;

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

import java.util.Map;

/**
 * 类名称：
 *
 * @author 李庆伟
 * @date 2024年03月04日 16:54
 */

@Component
public class RabbitMqListener {

    @RabbitListener(queues = "loss_queue")
    @RabbitHandler
    public void process(Message msg, Channel channel) {
        System.out.println("Rabbitmq Direct : " + msg);
        //设置的唯一id,可以用来处理重复消费
        String id = msg.getMessageProperties().getMessageId();

        //消息队列自身设置的唯一标识
        long tag = msg.getMessageProperties().getDeliveryTag();

        //int a = 1/0;

        try {
            //监听到要添加的用户信息
            String dataStr = new String(msg.getBody());
            Map<String,Object> addMap = JSON.parseObject(dataStr,Map.class);


            //先去redis中查询是否已经添加过了该用户
            //如果未添加[重复消费]
            if(true){ //如果未添加
                //添加用户业务
                //....add(addMap);
                //告诉队列该消息已经消费
                channel.basicAck(tag,false);
            } else {//如果已添加
                //告诉队列该消息已经消费
                channel.basicAck(tag,false);
            }

        } catch (Exception e) {
            try {
                //tag：消息序号
                //multiple：消息的标识，是否确认多条，false只确认当前一个消息收到，
                //                               true确认所有consumer获得的消息（成功消费，消息从队列中删除)
                //requeue：是否要退回到队列 true 将消息再次放到mq队列中，false是不把消息放到队列中
                channel.basicNack(tag, true, false);
                //channel.basicNack(tag, true, true); //如果能走到此处，这样会把消息在放到队列中，会在次被监听到，陷入死循环
                //channel.basicNack(tag, false, true); //如果能走到此处，这样会把消息在放到队列中，会在次被监听到，陷入死循环
                //channel.basicNack(tag, false, false); //如果能走到此处，如果是扇形交换机，其他消费者也会再次消费此信息

                //可以把添加用户业务数据保存起来
                //...
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

    }

}

6.App

package com.dev;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * 类名称：
 *
 * @author 李庆伟
 * @date 2024年03月04日 14:11
 */
@SpringBootApplication
public class App {

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

}