发布于 由admin

JAVA Redis 订阅消息丢失？PubSub 不持久化导致的架构风险

好的，让我们开始吧。

JAVA Redis 订阅消息丢失？PubSub 不持久化导致的架构风险

大家好，今天我们来深入探讨一个在使用 Redis Pub/Sub 功能时经常遇到的问题：消息丢失，以及它背后可能存在的架构风险。尤其是在 Java 环境下，理解这些风险并采取相应的措施至关重要。

Redis Pub/Sub 的基本概念和限制

Redis 的 Pub/Sub 是一种发布/订阅模式的消息传递机制。Publisher（发布者）将消息发送到特定的 Channel（频道），而 Subscriber（订阅者）可以订阅一个或多个频道，接收发布到这些频道的消息。

然而，Redis Pub/Sub 并非设计用于可靠的消息队列。它有以下几个关键限制：

  • 非持久化: Redis Pub/Sub 消息默认情况下不会被持久化到磁盘。一旦消息被发送到频道，如果当时没有订阅者在线，或者订阅者因为网络问题、客户端崩溃等原因无法及时接收消息，那么消息就会丢失。
  • At-Most-Once 语义: Redis Pub/Sub 提供的是 "至多一次" 的消息传递语义。这意味着消息可能会丢失，但不会被重复传递。
  • 没有消息确认机制: Redis Pub/Sub 没有内置的消息确认机制。发布者无法知道消息是否被成功传递给所有订阅者。
  • 订阅者掉线问题: 如果订阅者在接收消息期间掉线，当它重新连接时，它不会收到在掉线期间发布的消息。

这些限制使得 Redis Pub/Sub 不适合用于对数据可靠性要求高的场景，例如金融交易、订单处理等。

Java 环境下 Redis Pub/Sub 的使用

在 Java 中，我们可以使用 Jedis、Lettuce 等 Redis 客户端库来使用 Redis Pub/Sub 功能。

Jedis 示例：

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

public class RedisPubSubExample {

    public static void main(String[] args) {
        // 发布者
        new Thread(() -> {
            try (Jedis jedis = new Jedis("localhost", 6379)) {
                for (int i = 0; i < 10; i++) {
                    String message = "Message " + i;
                    jedis.publish("mychannel", message);
                    System.out.println("Published: " + message);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 订阅者
        new Thread(() -> {
            try (Jedis jedis = new Jedis("localhost", 6379)) {
                JedisPubSub jedisPubSub = new JedisPubSub() {
                    @Override
                    public void onMessage(String channel, String message) {
                        System.out.println("Received from " + channel + ": " + message);
                    }

                    @Override
                    public void onSubscribe(String channel, int subscribedChannels) {
                        System.out.println("Subscribed to " + channel);
                    }

                    @Override
                    public void onUnsubscribe(String channel, int subscribedChannels) {
                        System.out.println("Unsubscribed from " + channel);
                    }
                };
                jedis.subscribe(jedisPubSub, "mychannel");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

        try {
            Thread.sleep(5000); // 保证发布者和订阅者都运行一段时间
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Lettuce 示例：

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.pubsub.RedisPubSubAdapter;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;

public class LettucePubSubExample {

    public static void main(String[] args) throws InterruptedException {
        // 发布者
        new Thread(() -> {
            RedisClient redisClient = RedisClient.create(RedisURI.create("localhost", 6379));
            StatefulRedisConnection<String, String> connection = redisClient.connect();
            for (int i = 0; i < 10; i++) {
                String message = "Message " + i;
                connection.sync().publish("mychannel", message);
                System.out.println("Published: " + message);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            connection.close();
            redisClient.shutdown();
        }).start();

        // 订阅者
        new Thread(() -> {
            RedisClient redisClient = RedisClient.create(RedisURI.create("localhost", 6379));
            StatefulRedisPubSubConnection<String, String> pubSubConnection = redisClient.connectPubSub();
            pubSubConnection.addListener(new RedisPubSubAdapter<String, String>() {
                @Override
                public void message(String channel, String message) {
                    System.out.println("Received from " + channel + ": " + message);
                }

                @Override
                public void subscribed(String channel, long count) {
                    System.out.println("Subscribed to " + channel);
                }

                @Override
                public void unsubscribed(String channel, long count) {
                    System.out.println("Unsubscribed from " + channel);
                }
            });
            pubSubConnection.sync().subscribe("mychannel");

            try {
                Thread.sleep(5000); // 保持连接一段时间
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            pubSubConnection.close();
            redisClient.shutdown();
        }).start();

        Thread.sleep(5000); // 保证发布者和订阅者都运行一段时间
    }
}

这两个示例都演示了如何使用 Java Redis 客户端进行发布和订阅。 但是，它们都基于 Redis Pub/Sub 的非持久化特性，因此容易出现消息丢失的情况。

消息丢失的常见场景

  1. 订阅者未启动: 发布者在订阅者启动之前发布消息，消息会丢失。
  2. 订阅者掉线: 订阅者在接收消息期间掉线，掉线期间发布的消息会丢失。
  3. 网络问题: 网络不稳定导致订阅者无法及时接收消息，消息会丢失。
  4. Redis 服务器重启: Redis 服务器重启会导致所有未持久化的数据丢失，包括 Pub/Sub 消息。
  5. 客户端处理速度慢: 如果订阅者处理消息的速度跟不上发布者的发布速度，可能会导致消息积压，最终导致消息丢失或处理延迟。

架构风险：非持久化 Pub/Sub 的不当使用

在对数据可靠性有要求的场景中使用非持久化的 Redis Pub/Sub，会带来严重的架构风险。例如：

  • 数据不一致: 如果消息丢失导致部分业务逻辑未执行，可能会导致数据不一致。
  • 业务流程中断: 关键消息的丢失可能导致业务流程中断，影响用户体验。
  • 难以追踪问题: 消息丢失很难追踪和排查，增加了维护成本。
  • 系统不稳定: 大量消息丢失可能导致系统负载增加，最终导致系统崩溃。

替代方案：持久化消息队列

为了解决 Redis Pub/Sub 的消息丢失问题，我们需要使用持久化的消息队列。以下是一些常见的替代方案：

  • Redis Streams: Redis 5.0 引入了 Redis Streams，它提供了一种持久化的、可追加的消息队列。Streams 支持消息的持久化、消费者组、消息确认等功能，可以满足对数据可靠性要求高的场景。
  • RabbitMQ: RabbitMQ 是一个流行的开源消息队列，它支持多种消息传递模式，包括发布/订阅、点对点等。RabbitMQ 提供消息的持久化、消息确认、死信队列等功能，可以保证消息的可靠传递。
  • Kafka: Kafka 是一个高性能、分布式的消息队列，它主要用于处理大规模的数据流。Kafka 支持消息的持久化、分区、复制等功能，可以保证消息的可靠性和高吞吐量。
  • RocketMQ: 阿里开源的 RocketMQ 也是一个不错的选择，它在电商场景下经过了大规模的验证，性能和可靠性都比较高。

Redis Streams 的使用示例 (Java – Lettuce):

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.XAddArgs;
import io.lettuce.core.StreamMessage;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.XReadArgs;
import java.util.List;
import java.util.Map;

public class RedisStreamsExample {

    public static void main(String[] args) throws InterruptedException {
        RedisClient redisClient = RedisClient.create(RedisURI.create("localhost", 6379));
        StatefulRedisConnection<String, String> connection = redisClient.connect();
        RedisCommands<String, String> syncCommands = connection.sync();

        String streamKey = "mystream";
        String consumerGroup = "mygroup";
        String consumerName = "consumer1";

        // 创建 Stream 和 Consumer Group (如果不存在)
        try {
            syncCommands.xgroupCreate(streamKey, consumerGroup, "0-0", true);
        } catch (Exception e) {
            // Group already exists
        }

        // 发布消息
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                String message = "Message " + i;
                String id = syncCommands.xadd(streamKey, new XAddArgs().maxlen(1000), Map.of("data", message));
                System.out.println("Published message " + message + " with id " + id);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        // 消费消息
        new Thread(() -> {
            while (true) {
                XReadArgs xReadArgs = XReadArgs.Builder.block(2000).count(1);
                List<StreamMessage<String, String>> messages = syncCommands.xreadgroup(consumerGroup, consumerName, xReadArgs, streamKey, ">");

                if (messages != null && !messages.isEmpty()) {
                    for (StreamMessage<String, String> message : messages) {
                        String messageId = message.getId();
                        String data = message.getBody().get("data");
                        System.out.println("Consumer " + consumerName + " received message " + data + " with id " + messageId);

                        // 确认消息 (重要!)
                        syncCommands.xack(streamKey, consumerGroup, messageId);
                    }
                } else {
                    System.out.println("Consumer " + consumerName + " waiting for messages...");
                }

            }
        }).start();

        Thread.sleep(10000);
        connection.close();
        redisClient.shutdown();
    }
}

如何选择合适的方案

选择合适的方案需要考虑以下因素：

因素 Redis Streams RabbitMQ Kafka RocketMQ
数据可靠性 高 (持久化, 消息确认) 高 (持久化, 消息确认) 高 (持久化, 分区, 复制) 高 (持久化, 消息确认, 事务消息)
性能 中等 (单线程, 内存操作) 中等 (持久化开销) 高 (批量处理, 磁盘顺序写) 高 (批量处理, 磁盘顺序写)
复杂性 较低 (Redis 内置, 易于使用) 中等 (需要独立部署和配置) 高 (需要 ZooKeeper, 配置复杂) 中等 (需要 NameServer, 配置略复杂)
适用场景 简单、中等吞吐量的消息队列，需要事务支持的场景 复杂的消息路由，需要灵活的配置，可靠性要求高的场景 大规模数据流处理，高吞吐量，日志收集等场景 电商场景，高并发、低延迟、可靠性要求高的场景

应对策略和最佳实践

  1. 明确需求: 在选择 Redis Pub/Sub 之前，要明确业务需求，评估数据可靠性的要求。如果对数据可靠性要求高，应选择持久化的消息队列。
  2. 监控和告警: 建立完善的监控和告警机制，及时发现和处理消息丢失的问题。可以监控 Redis 的连接状态、消息积压情况等。
  3. 重试机制: 在订阅者端实现重试机制，当接收消息失败时，可以尝试重新订阅。
  4. 幂等性设计: 在订阅者端实现幂等性设计，确保即使消息被重复消费，也不会产生副作用。
  5. 数据备份: 定期备份 Redis 数据，以防止数据丢失。
  6. 谨慎使用: 除非你完全理解 Redis Pub/Sub 的限制，并且你的应用可以容忍消息丢失，否则不要使用它。

消息丢失并非末日，选择合适的工具至关重要

Redis Pub/Sub 并非万能的，它有其局限性。在选择消息传递机制时，要充分考虑业务需求和数据可靠性要求，选择合适的工具。当需要高可靠性时，请毫不犹豫地选择持久化的消息队列，例如 Redis Streams、RabbitMQ、Kafka 或 RocketMQ。

避免架构风险，保证系统稳定

理解 Redis Pub/Sub 的非持久化特性以及它可能带来的架构风险至关重要。通过选择合适的替代方案和采取相应的应对策略，我们可以避免消息丢失，保证系统的稳定性和可靠性。

选用合适的组件，保障消息的可靠传递

在对数据可靠性有要求的场景中，务必避免直接使用 Redis Pub/Sub。选择 Redis Streams、RabbitMQ、Kafka 等持久化消息队列，并结合合理的监控和重试机制，才能构建一个稳定可靠的消息传递系统。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注