渣打银行Java后端面试深度解析（二）：基础设施 — 分布式ID、Kafka可靠性与敏感数据安全

1
 0 | 00000000 00000000 00000000 00000000 00000000 0 | 00000 00000 | 000000000000
2
符号位(1) |              时间戳(41)                   | 机器ID(10) |  序列号(12)

1
public class SnowflakeIdGenerator {
2

3
    // 自定义纪元：2024-01-01 00:00:00 UTC 的毫秒时间戳
4
    private static final long EPOCH = 1704067200000L;
5

6
    // 各部分的位数
7
    private static final long WORKER_ID_BITS = 10L;
8
    private static final long SEQUENCE_BITS = 12L;
9

10
    // 最大值
11
    private static final long MAX_WORKER_ID = (1L << WORKER_ID_BITS) - 1;  // 1023
12
    private static final long SEQUENCE_MASK = (1L << SEQUENCE_BITS) - 1;   // 4095
13

14
    // 左移位数
15
    private static final long WORKER_ID_SHIFT = SEQUENCE_BITS;           // 12
16
    private static final long TIMESTAMP_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS; // 22
17

18
    private final long workerId;
19
    private long sequence = 0L;
20
    private long lastTimestamp = -1L;
21

22
    public SnowflakeIdGenerator(long workerId) {
23
        if (workerId < 0 || workerId > MAX_WORKER_ID) {
24
            throw new IllegalArgumentException("Worker ID must be between 0 and " + MAX_WORKER_ID);
25
        }
26
        this.workerId = workerId;
27
    }
28

29
    public synchronized long nextId() {
30
        long currentTimestamp = System.currentTimeMillis();
31

32
        if (currentTimestamp < lastTimestamp) {
33
            // 时钟回拨！这是雪花算法的致命问题
34
            throw new RuntimeException("Clock moved backwards. Refusing to generate id for "
35
                + (lastTimestamp - currentTimestamp) + " milliseconds");
36
        }
37

38
        if (currentTimestamp == lastTimestamp) {
39
            // 同一毫秒内，序列号递增
40
            sequence = (sequence + 1) & SEQUENCE_MASK;
41
            if (sequence == 0) {
42
                // 序列号用完了，等待下一毫秒
43
                currentTimestamp = waitNextMillis(lastTimestamp);
44
            }
45
        } else {
46
            // 不同毫秒，序列号重置为0
47
            sequence = 0L;
48
        }
49

50
        lastTimestamp = currentTimestamp;
51

52
        return ((currentTimestamp - EPOCH) << TIMESTAMP_SHIFT)
53
             | (workerId << WORKER_ID_SHIFT)
54
             | sequence;
55
    }
56

57
    private long waitNextMillis(long lastTimestamp) {
58
        long timestamp = System.currentTimeMillis();
59
        while (timestamp <= lastTimestamp) {
60
            timestamp = System.currentTimeMillis();
61
        }
62
        return timestamp;
63
    }
64
}

1
if (currentTimestamp < lastTimestamp) {
2
    throw new RuntimeException("Clock moved backwards");
3
}

1
if (currentTimestamp < lastTimestamp) {
2
    long offset = lastTimestamp - currentTimestamp;
3
    if (offset <= 5) { // 允许5ms以内的小回拨
4
        try {
5
            Thread.sleep(offset << 1); // 等待两倍的回拨时间
6
            currentTimestamp = System.currentTimeMillis();
7
            if (currentTimestamp < lastTimestamp) {
8
                throw new RuntimeException("Clock still backwards after wait");
9
            }
10
        } catch (InterruptedException e) {
11
            throw new RuntimeException(e);
12
        }
13
    } else {
14
        throw new RuntimeException("Clock moved backwards by " + offset + "ms");
15
    }
16
}

1
 0 | 时间戳(41) | 扩展位(2) | 机器ID(10) | 序列号(11)

1
private long epochPlus = 0L; // 回拨扩展位
2

3
public synchronized long nextId() {
4
    long currentTimestamp = System.currentTimeMillis();
5

6
    if (currentTimestamp < lastTimestamp) {
7
        epochPlus++;
8
        if (epochPlus > 3) {
9
            throw new RuntimeException("Too many clock backwards");
10
        }
11
    } else {
12
        epochPlus = 0;
13
    }
14
    // ... 使用 (currentTimestamp - EPOCH) | (epochPlus << 60) 来编码
15
}

1
/**
2
 * 银行定制化ID生成器
3
 *
4
 * 结构设计（64位）：
5
 * 符号位(1) + 日期偏移(13,支持22年) + 机构号(8,256个机构)
6
 * + 交易类型(4,16种交易) + 机器ID(6,64节点) + 序列号(12,4096/ms)
7
 *
8
 * 总计：1+13+8+4+6+12 = 44位（加上符号位45位）
9
 *
10
 * 剩余19位保留扩展使用
11
 */
12
public class BankIdGenerator {
13

14
    private static final long EPOCH = LocalDate.of(2024, 1, 1)
15
        .atStartOfDay(ZoneId.systemDefault()).toInstant().toEpochMilli();
16

17
    private static final long ORG_BITS = 8L;
18
    private static final long TX_TYPE_BITS = 4L;
19
    private static final long WORKER_BITS = 6L;
20
    private static final long SEQ_BITS = 12L;
21
    private static final long DATE_BITS = 13L;
22

23
    private static final long MAX_ORG = (1L << ORG_BITS) - 1;
24
    private static final long MAX_TX_TYPE = (1L << TX_TYPE_BITS) - 1;
25
    private static final long MAX_WORKER = (1L << WORKER_BITS) - 1;
26
    private static final long SEQ_MASK = (1L << SEQ_BITS) - 1;
27

28
    private final long orgCode;    // 机构号 0-255
29
    private final long txType;     // 交易类型 0-15
30
    private final long workerId;   // 节点ID 0-63
31
    private long sequence = 0L;
32
    private long lastTimestamp = -1L;
33

34
    public BankIdGenerator(long orgCode, long txType, long workerId) {
35
        this.orgCode = orgCode;
36
        this.txType = txType;
37
        this.workerId = workerId;
38
    }
39

40
    public synchronized long nextId() {
41
        long currentTimestamp = System.currentTimeMillis();
42
        long dayOffset = (currentTimestamp - EPOCH) / (24 * 3600 * 1000);
43

44
        if (currentTimestamp < lastTimestamp) {
45
            throw new RuntimeException("Clock moved backwards");
46
        }
47

48
        if (currentTimestamp == lastTimestamp) {
49
            sequence = (sequence + 1) & SEQ_MASK;
50
            if (sequence == 0) {
51
                currentTimestamp = waitNextMillis(lastTimestamp);
52
                dayOffset = (currentTimestamp - EPOCH) / (24 * 3600 * 1000);
53
            }
54
        } else {
55
            sequence = 0L;
56
        }
57

58
        lastTimestamp = currentTimestamp;
59

60
        return (dayOffset << (ORG_BITS + TX_TYPE_BITS + WORKER_BITS + SEQ_BITS))
61
             | (orgCode << (TX_TYPE_BITS + WORKER_BITS + SEQ_BITS))
62
             | (txType << (WORKER_BITS + SEQ_BITS))
63
             | (workerId << SEQ_BITS)
64
             | sequence;
65
    }
66

67
    /**
68
     * 反解ID，提取业务信息
69
     */
70
    public static Map<String, Object> decode(long id) {
71
        long seq = id & SEQ_MASK;
72
        long worker = (id >> SEQ_BITS) & MAX_WORKER;
73
        long tx = (id >> (WORKER_BITS + SEQ_BITS)) & MAX_TX_TYPE;
74
        long org = (id >> (TX_TYPE_BITS + WORKER_BITS + SEQ_BITS)) & MAX_ORG;
75
        long day = id >> (ORG_BITS + TX_TYPE_BITS + WORKER_BITS + SEQ_BITS);
76

77
        LocalDate date = LocalDate.ofEpochDay(day + LocalDate.of(2024, 1, 1).toEpochDay());
78

79
        Map<String, Object> info = new LinkedHashMap<>();
80
        info.put("date", date.toString());
81
        info.put("orgCode", org);
82
        info.put("txType", tx);
83
        info.put("workerId", worker);
84
        info.put("sequence", seq);
85
        return info;
86
    }
87

88
    private long waitNextMillis(long lastTs) {
89
        long ts = System.currentTimeMillis();
90
        while (ts <= lastTs) ts = System.currentTimeMillis();
91
        return ts;
92
    }
93
}

1
-- Leaf的数据库表设计
2
CREATE TABLE leaf_alloc (
3
    biz_tag VARCHAR(128) NOT NULL COMMENT '业务标签',
4
    max_id BIGINT NOT NULL DEFAULT 1 COMMENT '当前已分配的最大ID',
5
    step INT NOT NULL DEFAULT 1000 COMMENT '每次申请的号段大小',
6
    description VARCHAR(256) DEFAULT NULL,
7
    update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
8
    PRIMARY KEY (biz_tag)
9
) ENGINE=InnoDB;
10

11
-- 申请号段的SQL
12
UPDATE leaf_alloc SET max_id = max_id + step WHERE biz_tag = #{bizTag};

1
当前号段 [1, 1000] 使用到 900 时
2
    → 异步预申请下一个号段 [1001, 2000]
3
    → 当前号段用完时直接切换到新号段
4
    → 延迟接近于零

1
public class SegmentBuffer {
2
    private String key;              // 业务标签
3
    private Segment[] segments;      // 双Buffer
4
    private volatile int currentPos; // 当前使用的Buffer位置（0或1）
5

6
    public class Segment {
7
        private volatile long value;     // 当前已使用到的ID
8
        private long maxId;              // 号段最大值
9
        private long step;               // 号段步长
10
        private volatile boolean loaded; // 是否已加载完成
11
    }
12

13
    public long getId() {
14
        Segment current = segments[currentPos];
15
        long id = current.value.incrementAndGet();
16

17
        if (id >= current.maxId) {
18
            // 当前号段用完，切换到另一个Buffer
19
            currentPos = 1 - currentPos;
20
            Segment next = segments[currentPos];
21
            if (!next.loaded) {
22
                // 下一个号段还没加载好（极端情况），阻塞等待
23
                synchronized (this) {
24
                    while (!next.loaded) {
25
                        try { wait(10); } catch (InterruptedException e) { /* */ }
26
                    }
27
                }
28
            }
29
            id = next.value.incrementAndGet();
30
        }
31

32
        // 当前号段消耗到一定比例，触发异步预加载
33
        if ((id - current.value.get()) > current.step * 0.1) {
34
            // 异步加载另一个Buffer
35
            asyncLoadNextBuffer();
36
        }
37

38
        return id;
39
    }
40
}

1
Leaf启动 → 向ZK注册临时节点 → 获取workerId → 定期上报状态

1
生产者(Producer) → Kafka Broker(服务端) → 消费者(Consumer)
2
    ① 发送失败           ② 存储故障           ③ 消费失败

1
// 同步发送，等待Broker确认
2
try {
3
    RecordMetadata metadata = producer.send(record).get(); // 阻塞等待
4
    log.info("Message sent to partition={}, offset={}", metadata.partition(), metadata.offset());
5
} catch (ExecutionException e) {
6
    // 发送失败，需要处理（重试或记录）
7
    log.error("Failed to send message", e);
8
}

1
producer.send(record, (metadata, exception) -> {
2
    if (exception != null) {
3
        // 发送失败：记录到本地重试表，后续补偿发送
4
        log.error("Failed to send message to topic={}", record.topic(), exception);
5
        saveToRetryTable(record); // 持久化到本地数据库
6
    } else {
7
        log.info("Message sent: topic={}, partition={}, offset={}",
8
            metadata.topic(), metadata.partition(), metadata.offset());
9
    }
10
});

1
# Producer配置
2
retries=3                          # 重试次数
3
retry.backoff.ms=100               # 重试间隔
4
delivery.timeout.ms=120000         # 发送超时时间
5
enable.idempotence=true            # 开启幂等性（关键！）
6
max.in.flight.requests.per.connection=5  # 幂等模式下最多允许5个未确认请求

1
# acks参数决定Producer需要多少个Broker副本确认才算发送成功
2
acks=0    # 不等待确认，性能最高但可能丢消息（绝对不能用于银行场景）
3
acks=1    # 等待Leader确认，Leader挂了可能丢消息（不推荐银行场景）
4
acks=all  # 等待所有ISR副本确认（银行场景必须用这个！）

1
# Broker端配置
2
default.replication.factor=3           # 副本因子：每个分区3个副本
3
min.insync.replicas=2                  # 最少2个副本同步成功才返回确认
4
unclean.leader.election.enable=false   # 禁止非ISR副本成为Leader（防止数据丢失）

1
场景：3个Broker，replication.factor=3，min.insync.replicas=2
2

3
正常情况：
4
  Broker1(Leader) ← 写入成功
5
  Broker2(Follower) ← 同步成功  → 返回Producer成功（ISR有2个同步成功）
6
  Broker3(Follower) ← 同步成功
7

8
Broker1宕机：
9
  Broker2(成为新Leader) ← 数据完整
10
  Broker3(Follower) ← 数据完整
11
  → 不丢数据
12

13
Broker2和Broker3同时宕机：
14
  ISR中同步副本 < min.insync.replicas(2)
15
  → Producer发送失败（返回NotEnoughReplicasException）
16
  → 不会丢数据（宁可不可用也不能丢数据）

1
# Consumer配置
2
enable.auto.commit=false            # 关闭自动提交！
3
max.poll.records=500                # 每次最多拉取500条
4
session.timeout.ms=30000            # 会话超时

1
@KafkaListener(topics = "bank-transactions", groupId = "tx-processor")
2
public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
3
    try {
4
        for (ConsumerRecord<String, String> record : records) {
5
            // 业务处理
6
            processTransaction(record.value());
7
        }
8
        // 所有消息处理成功后，手动提交offset
9
        ack.acknowledge();
10
    } catch (Exception e) {
11
        // 处理失败，不提交offset，消息会被重新消费
12
        log.error("Failed to process messages", e);
13
        throw e; // 抛出异常，触发Consumer rebalance
14
    }
15
}

1
@Component
2
public class TransactionConsumer {
3

4
    @Autowired
5
    private TransactionLogRepository txLogRepo;
6

7
    public void processTransaction(String message) {
8
        TransactionMessage tx = JsonUtil.fromJson(message, TransactionMessage.class);
9

10
        // 幂等检查：根据消息中的唯一业务ID判断是否已处理
11
        if (txLogRepo.existsByExternalId(tx.getExternalId())) {
12
            log.info("Transaction already processed: {}", tx.getExternalId());
13
            return; // 已处理过，跳过
14
        }
15

16
        // 处理交易
17
        doProcessTransaction(tx);
18

19
        // 记录处理标记（与业务操作在同一事务中）
20
        txLogRepo.save(new TransactionLog(tx.getExternalId(), "PROCESSED"));
21
    }
22
}

1
// Producer配置
2
enable.idempotence=true   // 开启幂等
3
transactional.id=bank-tx-producer-001  // 事务ID（全局唯一）
4

5
// 代码
6
producer.initTransactions();
7
try {
8
    producer.beginTransaction();
9
    producer.send(new ProducerRecord<>("topic1", key1, value1));
10
    producer.send(new ProducerRecord<>("topic2", key2, value2));
11
    producer.commitTransaction();
12
} catch (ProducerFencedException e) {
13
    producer.close(); // 另一个Producer使用了相同的transactional.id
14
} catch (KafkaException e) {
15
    producer.abortTransaction();
16
}

1
# Consumer需要设置隔离级别为read_committed
2
isolation.level=read_committed

1
# ===== 生产端 =====
2
acks=all
3
retries=3
4
retry.backoff.ms=100
5
enable.idempotence=true
6
max.in.flight.requests.per.connection=5
7
delivery.timeout.ms=120000
8
max.request.size=10485760  # 10MB
9

10
# ===== Broker端 =====
11
default.replication.factor=3
12
min.insync.replicas=2
13
unclean.leader.election.enable=false
14
log.retention.hours=168  # 消息保留7天（银行可能更长）
15
log.retention.bytes=-1   # 不按大小删除
16

17
# ===== 消费端 =====
18
enable.auto.commit=false
19
isolation.level=read_committed
20
max.poll.records=500
21
session.timeout.ms=30000

1
@Component
2
public class KafkaHealthMonitor {
3

4
    // 监控Producer发送失败率
5
    @Scheduled(fixedDelay = 60000)
6
    public void checkProducerMetrics() {
7
        Map<MetricName, ? extends Metric> metrics = producer.metrics();
8
        double failedSendRate = getMetricValue(metrics, "failed-send-rate");
9
        if (failedSendRate > 0.01) { // 失败率超过1%
10
            alertService.sendAlert("Kafka Producer发送失败率异常: " + failedSendRate);
11
        }
12
    }
13

14
    // 监控Consumer Lag（消费延迟）
15
    @Scheduled(fixedDelay = 30000)
16
    public void checkConsumerLag() {
17
        Map<String, Map<Integer, Long>> lags = kafkaAdmin.getConsumerLag("bank-consumer-group");
18
        for (Map.Entry<String, Map<Integer, Long>> entry : lags.entrySet()) {
19
            long totalLag = entry.getValue().values().stream().mapToLong(Long::longValue).sum();
20
            if (totalLag > 10000) {
21
                alertService.sendAlert("Consumer Lag过大: topic=" + entry.getKey() + ", lag=" + totalLag);
22
            }
23
        }
24
    }
25
}

1
// 使用Bouncy Castle库实现SM4
2
import org.bouncycastle.jce.provider.BouncyCastleProvider;
3

4
public class SM4Encryptor {
5

6
    static {
7
        Security.addProvider(new BouncyCastleProvider());
8
    }
9

10
    private static final String ALGORITHM = "SM4/GCM/NoPadding";
11

12
    public String encrypt(String plaintext, byte[] key) {
13
        try {
14
            Cipher cipher = Cipher.getInstance(ALGORITHM, "BC");
15
            byte[] iv = new byte[16];
16
            SecureRandom.getInstanceStrong().nextBytes(iv);
17
            GCMParameterSpec spec = new GCMParameterSpec(128, iv);
18

19
            SecretKeySpec keySpec = new SecretKeySpec(key, "SM4");
20
            cipher.init(Cipher.ENCRYPT_MODE, keySpec, spec);
21

22
            byte[] ciphertext = cipher.doFinal(plaintext.getBytes(StandardCharsets.UTF_8));
23

24
            byte[] combined = new byte[iv.length + ciphertext.length];
25
            System.arraycopy(iv, 0, combined, 0, iv.length);
26
            System.arraycopy(ciphertext, 0, combined, iv.length, ciphertext.length);
27

28
            return Base64.getEncoder().encodeToString(combined);
29
        } catch (Exception e) {
30
            throw new SecurityException("SM4 encryption failed", e);
31
        }
32
    }
33

34
    public String decrypt(String ciphertext, byte[] key) {
35
        try {
36
            byte[] combined = Base64.getDecoder().decode(ciphertext);
37
            byte[] iv = Arrays.copyOfRange(combined, 0, 16);
38
            byte[] encrypted = Arrays.copyOfRange(combined, 16, combined.length);
39

40
            Cipher cipher = Cipher.getInstance(ALGORITHM, "BC");
41
            SecretKeySpec keySpec = new SecretKeySpec(key, "SM4");
42
            cipher.init(Cipher.DECRYPT_MODE, keySpec, new GCMParameterSpec(128, iv));
43

44
            return new String(cipher.doFinal(encrypted), StandardCharsets.UTF_8);
45
        } catch (Exception e) {
46
            throw new SecurityException("SM4 decryption failed", e);
47
        }
48
    }
49
}

1
@Component
2
public class DataMasker {
3

4
    /**
5
     * 手机号脱敏：138****1234
6
     */
7
    public String maskPhone(String phone) {
8
        if (phone == null || phone.length() < 7) return phone;
9
        return phone.substring(0, 3) + "****" + phone.substring(phone.length() - 4);
10
    }
11

12
    /**
13
     * 身份证脱敏：110***********1234
14
     */
15
    public String maskIdCard(String idCard) {
16
        if (idCard == null || idCard.length() < 8) return idCard;
17
        return idCard.substring(0, 3) + "*".repeat(idCard.length() - 7) + idCard.substring(idCard.length() - 4);
18
    }
19

20
    /**
21
     * 银行卡号脱敏：6222 **** **** 1234
22
     */
23
    public String maskBankCard(String cardNo) {
24
        if (cardNo == null || cardNo.length() < 8) return cardNo;
25
        return cardNo.substring(0, 4) + " **** **** " + cardNo.substring(cardNo.length() - 4);
26
    }
27

28
    /**
29
     * 姓名脱敏：张*明、欧阳**（保留姓氏）
30
     */
31
    public String maskName(String name) {
32
        if (name == null || name.length() < 2) return name;
33
        if (name.length() == 2) {
34
            return name.charAt(0) + "*";
35
        }
36
        return name.charAt(0) + "*".repeat(name.length() - 2) + name.charAt(name.length() - 1);
37
    }
38
}

1
public class UserVO {
2
    @MaskType(MaskType.PHONE)
3
    private String phone;
4

5
    @MaskType(MaskType.ID_CARD)
6
    private String idCard;
7

8
    @MaskType(MaskType.BANK_CARD)
9
    private String bankCard;
10
}
11

12
// 自定义Jackson序列化器
13
public class MaskingSerializer extends JsonSerializer<String> {
14

15
    @Override
16
    public void serialize(String value, JsonGenerator gen, SerializerProvider provider) throws IOException {
17
        MaskType maskType = // 从注解获取
18
        gen.writeString(DataMasker.mask(value, maskType));
19
    }
20
}

1
SELECT CONCAT(LEFT(phone, 3), '****', RIGHT(phone, 4)) AS phone
2
FROM users WHERE id = ?

1
密钥层级：
2
  主密钥(MK) → 存储在HSM中
3
    → 加密 → 数据密钥1(DEK1) → 加密 → 用户表手机号字段
4
    → 加密 → 数据密钥2(DEK2) → 加密 → 用户表身份证字段
5
    → 加密 → 数据密钥3(DEK3) → 加密 → 交易表金额字段