使用Kafka实现Java异步更新通知解决Redis与MySQL数据不一致

<dependencies>
    <!-- MySQL连接驱动 -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.23</version>
    </dependency>

    <!-- Jedis：Java连接Redis的客户端库 -->
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>3.7.0</version>
    </dependency>

    <!-- Kafka客户端 -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.8.1</version>
    </dependency>
</dependencies>

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Properties;
import java.util.Timer;
import java.util.TimerTask;

public class AsyncInventoryNotifier {

    // Redis连接信息
    private static final String REDIS_HOST = "localhost";
    private static final int REDIS_PORT = 6379;
    private static final int REDIS_DB = 0;

    // MySQL连接信息
    private static final String MYSQL_URL = "jdbc:mysql://localhost:3306/ecommerce";
    private static final String MYSQL_USER = "user";
    private static final String MYSQL_PASSWORD = "password";

    // Kafka连接信息
    private static final String KAFKA_BROKER = "localhost:9092";
    private static final String KAFKA_TOPIC = "inventory_updates";

    public static void main(String[] args) {
        // 创建定时任务
        Timer timer = new Timer();
        timer.schedule(new InventoryNotifierTask(), 0, 30 * 60 * 1000); // 每30分钟执行一次
    }

    static class InventoryNotifierTask extends TimerTask {
        @Override
        public void run() {
            System.out.println("Starting async inventory notification...");

            try {
                // 连接Redis
                Jedis jedis = new Jedis(REDIS_HOST, REDIS_PORT);
                jedis.select(REDIS_DB);

                // 连接MySQL
                Connection mysqlConnection = DriverManager.getConnection(MYSQL_URL, MYSQL_USER, MYSQL_PASSWORD);

                // 连接Kafka
                Properties kafkaProps = new Properties();
                kafkaProps.put("bootstrap.servers", KAFKA_BROKER);
                kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
                kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
                KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps);

                // 查询所有商品ID
                PreparedStatement preparedStatement = mysqlConnection.prepareStatement("SELECT id FROM products");
                ResultSet resultSet = preparedStatement.executeQuery();

                while (resultSet.next()) {
                    int productId = resultSet.getInt("id");

                    // 从Redis获取缓存库存
                    int redisStock = Integer.parseInt(jedis.get("product:" + productId + ":stock"));

                    // 从MySQL获取实际库存
                    PreparedStatement stockStatement = mysqlConnection.prepareStatement("SELECT stock FROM products WHERE id = ?");
                    stockStatement.setInt(1, productId);
                    ResultSet stockResultSet = stockStatement.executeQuery();
                    int mysqlStock = 0;
                    if (stockResultSet.next()) {
                        mysqlStock = stockResultSet.getInt("stock");
                    }

                    // 检测库存一致性
                    if (redisStock != mysqlStock) {
                        System.out.println("Inventory inconsistency detected for product " + productId +
                                ". Redis: " + redisStock + ", MySQL: " + mysqlStock);

                        // 发送异步通知到Kafka
                        ProducerRecord<String, String> record = new ProducerRecord<>(KAFKA_TOPIC, String.valueOf(productId), "inventory_inconsistency");
                        producer.send(record);
                        System.out.println("Async notification sent to Kafka.");
                    }
                }

                // 关闭连接
                jedis.close();
                mysqlConnection.close();
                producer.close();

            } catch (SQLException e) {
                System.err.println("Error during async inventory notification: " + e.getMessage());
            }
        }
    }
}

Jedis jedis = new Jedis(REDIS_HOST, REDIS_PORT);
jedis.select(REDIS_DB);

Connection mysqlConnection = DriverManager.getConnection(MYSQL_URL, MYSQL_USER, MYSQL_PASSWORD);

Properties kafkaProps = new Properties();
kafkaProps.put("bootstrap.servers", KAFKA_BROKER);
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps);

PreparedStatement preparedStatement = mysqlConnection.prepareStatement("SELECT id FROM products");
ResultSet resultSet = preparedStatement.executeQuery();

while (resultSet.next()) {
    int productId = resultSet.getInt("id");

    int redisStock = Integer.parseInt(jedis.get("product:" + productId + ":stock"));

    PreparedStatement stockStatement = mysqlConnection.prepareStatement("SELECT stock FROM products WHERE id = ?");
    stockStatement.setInt(1, productId);
    ResultSet stockResultSet = stockStatement.executeQuery();
    int mysqlStock = 0;
    if (stockResultSet.next()) {
        mysqlStock = stockResultSet.getInt("stock");
    }

    if (redisStock != mysqlStock) {
        // 发送异步通知到Kafka
        ProducerRecord<String, String> record = new ProducerRecord<>(KAFKA_TOPIC, String.valueOf(productId), "inventory_inconsistency");
        producer.send(record

3. **异步通知：**
```java
ProducerRecord<String, String> record = new ProducerRecord<>(KAFKA_TOPIC, String.valueOf(productId), "inventory_inconsistency");
producer.send(record);
System.out.println("Async notification sent to Kafka.");

jedis.close();
mysqlConnection.close();
producer.close();