面试题：微服务架构下跨服务数据库交互的优化策略

优化跨服务数据库交互策略

分布式事务管理
- 两阶段提交（2PC）：协调者先向所有参与者发送准备消息，参与者执行事务操作但不提交。若所有参与者准备成功，协调者发送提交消息，参与者正式提交事务；若有任何一个参与者准备失败，协调者发送回滚消息。
- 三阶段提交（3PC）：在2PC基础上增加预询问阶段，协调者先询问参与者能否执行事务，参与者回复可执行能力。然后进入准备阶段和提交阶段，减少单点故障和阻塞问题。
数据缓存
- 在服务层使用缓存（如Redis），对于读多写少的数据，先从缓存读取，命中则直接返回，减少数据库压力。写操作时同时更新缓存和数据库，可采用缓存更新策略（如写后失效、写时更新等）。
数据库连接池
- 每个服务维护自己的数据库连接池，减少连接创建和销毁的开销。配置合适的连接池参数，如最大连接数、最小连接数、连接超时时间等。
异步处理
- 将一些非关键的数据库操作异步化，如使用消息队列（如Kafka、RabbitMQ）。服务将数据库操作相关消息发送到队列，由专门的消费者从队列中取出消息并执行数据库操作，避免服务等待。

实现方式

分布式事务管理
- 2PC实现：可使用开源框架如Atomikos，通过XA协议实现分布式事务。在代码中定义事务管理器，配置数据源等相关信息。例如，在Java Spring Boot项目中，引入Atomikos依赖，配置XA数据源和事务管理器，在需要事务的方法上添加@Transactional注解。
- 3PC实现：相对复杂，目前较少有直接可用的成熟框架，可能需要自行基于消息传递机制实现各个阶段的逻辑，确保参与者和协调者之间的状态同步和消息可靠传递。
数据缓存
- 以Redis为例，在服务中引入Redis客户端依赖（如Jedis、Lettuce）。在读取数据方法中，先尝试从Redis获取数据，若不存在则从数据库读取并写入Redis。写数据时，先更新数据库，再更新Redis缓存。如在Java中：

import redis.clients.jedis.Jedis;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class DataAccess {
    private static final String REDIS_HOST = "localhost";
    private static final int REDIS_PORT = 6379;
    private static final String DB_URL = "jdbc:mysql://localhost:3306/mydb";
    private static final String DB_USER = "user";
    private static final String DB_PASSWORD = "password";

    public String getData(String key) {
        try (Jedis jedis = new Jedis(REDIS_HOST, REDIS_PORT)) {
            String value = jedis.get(key);
            if (value != null) {
                return value;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        try (Connection conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASSWORD)) {
            String sql = "SELECT data FROM my_table WHERE key =?";
            try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
                pstmt.setString(1, key);
                try (ResultSet rs = pstmt.executeQuery()) {
                    if (rs.next()) {
                        String data = rs.getString("data");
                        try (Jedis jedis = new Jedis(REDIS_HOST, REDIS_PORT)) {
                            jedis.set(key, data);
                        }
                        return data;
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public void setData(String key, String value) {
        try (Connection conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASSWORD)) {
            String sql = "INSERT INTO my_table (key, data) VALUES (?,?) ON DUPLICATE KEY UPDATE data =?";
            try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
                pstmt.setString(1, key);
                pstmt.setString(2, value);
                pstmt.setString(3, value);
                pstmt.executeUpdate();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        try (Jedis jedis = new Jedis(REDIS_HOST, REDIS_PORT)) {
            jedis.set(key, value);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

数据库连接池
- 在Java中，可使用HikariCP连接池。在项目配置文件（如Spring Boot的application.properties）中配置相关参数：

spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/mydb
spring.datasource.username=user
spring.datasource.password=password
spring.datasource.hikari.maximum-pool-size=10
spring.datasource.hikari.minimum-idle=5
spring.datasource.hikari.connection-timeout=30000

异步处理
- 以Kafka为例，在生产者服务中引入Kafka生产者依赖，创建生产者配置并发送消息。在消费者服务中引入Kafka消费者依赖，创建消费者配置并消费消息执行数据库操作。例如在Java中： 生产者：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerExample {
    private static final String TOPIC = "database - operations";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            String message = "INSERT INTO my_table (column1, column2) VALUES ('value1', 'value2')";
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, message);
            producer.send(record);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

消费者：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerExample {
    private static final String TOPIC = "database - operations";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String DB_URL = "jdbc:mysql://localhost:3306/mydb";
    private static final String DB_USER = "user";
    private static final String DB_PASSWORD = "password";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "database - operation - group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Arrays.asList(TOPIC));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    String sql = record.value();
                    try (Connection conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASSWORD)) {
                        try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
                            pstmt.executeUpdate();
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

监控策略

分布式事务监控
- 使用监控工具如SkyWalking，它可以跟踪分布式事务的执行流程，记录事务的开始、准备、提交或回滚时间，以及各个参与者的状态。通过分析这些数据，可发现事务执行过程中的性能瓶颈和异常情况，如长时间等待某个参与者响应。
缓存监控
- 利用Redis自带的监控命令（如INFO命令）或第三方监控工具（如Prometheus + Grafana），监控缓存的命中率、内存使用情况、请求频率等指标。命中率低可能表示缓存策略不合理，内存使用过高可能需要调整缓存淘汰策略。
数据库连接池监控
- HikariCP等连接池提供了一些内置的监控指标，如活跃连接数、空闲连接数、等待连接的线程数等。可通过JMX（Java Management Extensions）将这些指标暴露出来，使用工具如JConsole或与Prometheus集成，实时监控连接池的健康状态，避免连接池耗尽导致服务不可用。
异步处理监控
- 对于消息队列，使用其自带的监控工具（如Kafka的Kafka Manager）监控消息的发送速率、消费速率、堆积情况等。消费速率过慢或消息堆积可能表示消费者处理能力不足，需要调整消费者的并发数或优化数据库操作逻辑。

星途面试题库

面试题：微服务架构下跨服务数据库交互的优化策略

知识考点

面试题答案

优化跨服务数据库交互策略

实现方式

监控策略