rabbitMQ解决消息丢失-生产者发布确认

灰太狼 2023-01-23 07:57 52阅读 0赞

maven配置

<?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 https://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.5.0</version>
            <relativePath/> 
        </parent>
        <groupId>com.example</groupId>
        <artifactId>SpringbootTest</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>SpringbootTest</name>
        <description>Demo project for Spring Boot</description>
        <properties>
            <java.version>1.8</java.version>
        </properties>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter</artifactId>
            </dependency>
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.10.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-devtools</artifactId>
                <scope>runtime</scope>
                <optional>true</optional>
            </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>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                    <configuration>
                        <excludes>
                            <exclude>
                                <groupId>org.projectlombok</groupId>
                                <artifactId>lombok</artifactId>
                            </exclude>
                        </excludes>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
    </project>

工具类

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class MqChanneUtils {
        public static Channel getChannel() throws Exception{
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setVirtualHost("/");
            factory.setConnectionTimeout(20000);
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            return channel;
        }
    }

同步确认机制：开启信道确认模式，信道使用waitForConfirms应答说明生产者消息信道已收到。

public class Product01 {
        @SneakyThrows
        public static void main(String[] args) {
            Channel channel = MqChanneUtils.getChannel();
            channel.confirmSelect();//开启发布确认模式
            String message ="消息";
            long starttime = System.currentTimeMillis();
            for (int i = 0;i<1000;i++){
                channel.basicPublish("","test02",null,(message+i).getBytes(StandardCharsets.UTF_8));
                boolean flag = channel.waitForConfirms();//通道答复
                if(flag){
                    System.out.println("发布消息"+i+"成功");            }
            }
            long endtime = System.currentTimeMillis();
            System.out.println("时间差>>>>>>>"+(endtime-starttime)+"ms");
    
        }
    }

异步确认机制：开启信道确认模式。通过异步回调接口来接收信道给的确认消息。

public class Product03 {
        @SneakyThrows
        public static void main(String[] args) {
            Channel channel = MqChanneUtils.getChannel();
            channel.confirmSelect();
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new
                    ConcurrentSkipListMap<>();
            //这是确认发送成功的回调处理
            ConfirmCallback confirmCallback = (seq,flag )->{
                if (flag){
                    //如果是ture，确认当前消息及之前序列消息已经发送成功、
                    ConcurrentNavigableMap<Long, String> longStringConcurrentNavigableMap = outstandingConfirms.headMap(seq);
                    longStringConcurrentNavigableMap.clear();
                }else{
                    outstandingConfirms.remove(seq);
                }
            };
            //这是暂未确认的
            ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {
                String message = outstandingConfirms.get(sequenceNumber);
                System.out.println(" 发布的消息"+message+" 未被确认，序列号"+sequenceNumber);
            };
    
            channel.addConfirmListener(confirmCallback,null);
            long start = System.currentTimeMillis();
    
            for (int i = 0; i < 1000; i++) {
                String message = " 消息" + i;
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", queueName, null, message.getBytes());
            }
            long end = System.currentTimeMillis();
            System.out.println(" 发布" + 1000 + " 个异步确认消息, 耗时" + (end - start) +
                    "ms");
        }
    
    }