本文目录导读：
我来系统性地介绍Java分布式系统中数据组合的几种核心模式及其组合方式。
基本组合模式
1 数据分区模式
// 一致性哈希实现分片
public class ConsistentHashRouter<T> {
private final HashFunction hashFunction;
private final TreeMap<Long, VirtualNode<T>> ring = new TreeMap<>();
public T route(String key) {
Long hash = hashFunction.hash(key);
SortedMap<Long, VirtualNode<T>> tailMap = ring.tailMap(hash);
Long nodeHash = tailMap.isEmpty() ? ring.firstKey() : tailMap.firstKey();
return ring.get(nodeHash).getPhysicalNode();
}
}
// 范围分区
public class RangePartitioner {
private final NavigableMap<Long, DataNode> partitions = new ConcurrentSkipListMap<>();
public DataNode getPartition(Long key) {
return partitions.ceilingEntry(key).getValue();
}
}
2 数据复制模式
// 主从复制
public class MasterSlaveReplicator {
private final DataNode master;
private final List<DataNode> slaves;
@Transactional
public void write(Data data) {
master.write(data);
// 异步复制到从节点
CompletableFuture.allOf(
slaves.stream()
.map(slave -> CompletableFuture.runAsync(() -> slave.sync(data)))
.toArray(CompletableFuture[]::new)
).join();
}
public Data read(String key) {
// 读写分离
DataNode node = selectSlaveByLoad();
return node.read(key);
}
}
// 多层复制
public class MultiLayerReplication {
private final ReplicationGroup globalGroup;
private final ReplicationGroup regionalGroup;
private final ReplicationGroup localGroup;
public void replicate(Data data) {
// 全局层
globalGroup.broadcast(data);
// 区域层
regionalGroup.quorumReplicate(data, 3);
// 本地层
localGroup.asyncReplicate(data);
}
}
缓存组合模式
1 多级缓存组合
public class MultiLevelCache<K, V> {
private final Cache<K, V> l1Cache; // 本地缓存
private final Cache<K, V> l2Cache; // 分布式缓存(Redis)
private final Database<K, V> db; // 数据库
public V get(K key) {
// L1 缓存查询
V value = l1Cache.get(key);
if (value != null) return value;
// L2 缓存查询
value = l2Cache.get(key);
if (value != null) {
l1Cache.put(key, value);
return value;
}
// 数据库查询
value = db.query(key);
if (value != null) {
l1Cache.put(key, value);
l2Cache.put(key, value);
}
return value;
}
@Transactional
public void put(K key, V value) {
// 先更新数据库
db.update(key, value);
// 异步失效缓存
CompletableFuture.runAsync(() -> {
l1Cache.invalidate(key);
l2Cache.invalidate(key);
});
}
}
2 缓存穿透/击穿/雪崩防护
public class CacheProtection<K, V> {
private final Cache<K, V> cache;
private final Database<K, V> db;
private final BloomFilter<K> bloomFilter;
// 布隆过滤器防止缓存穿透
public V getWithBloomFilter(K key) {
if (!bloomFilter.mightContain(key)) {
return null; // 肯定不存在
}
return getWithProtection(key);
}
// 互斥锁防止缓存击穿
public V getWithProtection(K key) {
V value = cache.get(key);
if (value != null) return value;
// 互斥锁
String lockKey = "lock:" + key;
if (redisSetNx(lockKey, "1", 60)) {
try {
value = db.query(key);
if (value != null) {
cache.put(key, value, 300); // 设置过期时间
}
} finally {
redisDel(lockKey);
}
} else {
// 等待重试
Thread.sleep(50);
return getWithProtection(key);
}
return value;
}
}
数据聚合模式
1 MapReduce聚合
public class DistributedAggregator {
// Map阶段
public interface Mapper<T, K, V> {
void map(T input, Context<K, V> context);
}
// Reduce阶段
public interface Reducer<K, V, R> {
R reduce(K key, List<V> values);
}
// Shuffle阶段
public <K, V> Map<K, List<V>> shuffle(List<Pair<K, V>> mappedData) {
return mappedData.stream()
.collect(Collectors.groupingBy(
Pair::getKey,
Collectors.mapping(Pair::getValue, Collectors.toList())
));
}
// 执行聚合
public <T, K, V, R> Map<K, R> aggregate(
List<T> data,
Mapper<T, K, V> mapper,
Reducer<K, V, R> reducer) {
// 1. Map阶段
List<Pair<K, V>> mappedData = data.parallelStream()
.flatMap(item -> {
List<Pair<K, V>> results = new ArrayList<>();
mapper.map(item, new Context<K, V>() {
@Override
public void emit(K key, V value) {
results.add(new Pair<>(key, value));
}
});
return results.stream();
})
.collect(Collectors.toList());
// 2. Shuffle阶段
Map<K, List<V>> grouped = shuffle(mappedData);
// 3. Reduce阶段
Map<K, R> result = grouped.entrySet().parallelStream()
.collect(Collectors.toMap(
Map.Entry::getKey,
entry -> reducer.reduce(entry.getKey(), entry.getValue())
));
return result;
}
}
2 多维度聚合
public class MultiDimensionAggregator {
@Data
public static class AggregationKey {
private String dimension;
private String dimensionValue;
private List<String> tags;
// equals和hashCode基于所有字段
}
// 时间窗口聚合
public Map<AggregationKey, Double> timeWindowAggregate(
List<DataPoint> points, TimeWindow window) {
// 按时间窗口分组
return points.stream()
.collect(Collectors.groupingBy(
point -> new AggregationKey(
"time",
window.getWindowKey(point.getTimestamp()),
point.getTags()
),
Collectors.averagingDouble(DataPoint::getValue)
));
}
// 多维度OLAP聚合
public class OLAQuery {
private List<String> groupByDimensions;
private List<AggregationFunction> aggregations;
private Map<String, Object> filters;
public ResultSet execute() {
// 生成执行计划
QueryPlan plan = optimizeQuery();
// 分布式执行
return plan.execute();
}
}
}
最终一致性组合模式
1 事件驱动组合
public class EventualConsistencyModel {
private final EventBus eventBus;
private final Map<String, Saga> sagas;
@Data
public static class Event {
private String id;
private String type;
private Object data;
private long timestamp;
}
// Saga模式保证最终一致性
public class Saga {
private List<Step> steps;
private Stack<Step> executedSteps;
@Transactional
public void execute(Event event) {
try {
for (Step step : steps) {
step.execute(event);
executedSteps.push(step);
}
} catch (Exception e) {
// 回滚已执行的步骤
while (!executedSteps.isEmpty()) {
executedSteps.pop().compensate(event);
}
throw e;
}
}
}
}
2 CQRS模式
public class CQRSModel {
private final EventStore eventStore;
private final CommandHandler commandHandler;
private final QueryHandler queryHandler;
private final ViewUpdater viewUpdater;
// 命令端
public class CommandHandler {
@Transactional
public void handle(Command command) {
// 验证命令
validate(command);
// 生成事件
Event event = command.toEvent();
// 保存事件
eventStore.save(event);
// 发布事件
eventBus.publish(event);
}
}
// 查询端
public class QueryHandler {
private final MaterializedView view;
public Object handle(Query query) {
// 从物化视图查询
return view.query(query);
}
// 异步更新视图
public void updateView(Event event) {
CompletableFuture.runAsync(() -> {
// 应用事件到物化视图
view.apply(event);
});
}
}
}
组合优化模式
1 读写分离组合
public class ReadWriteSplitting {
private final WriteNode writeNode;
private final List<ReadNode> readNodes;
private final LoadBalancer loadBalancer;
// 写操作
@Transactional
public void write(Object data) {
// 写入主库
writeNode.write(data);
// 更新缓存
invalidateCache(data);
}
// 读操作
public Object read(String key) {
// 选择读取节点
ReadNode node = loadBalancer.select(readNodes);
Object data = node.read(key);
if (data == null) {
// 延迟读取主库
data = writeNode.read(key);
}
return data;
}
// 延迟读取策略
public class DelayedReadStrategy {
private long delayTime = 100; // 毫秒
public Object read(String key) {
// 等待延迟
Thread.sleep(delayTime);
// 从主库读取最新数据
return writeNode.read(key);
}
}
}
2 负载均衡组合
public class LoadBalancedDistribution {
private final List<Node> nodes;
private final LoadBalanceStrategy strategy;
public interface LoadBalanceStrategy {
Node select(List<Node> nodes, String key);
}
// 一致性哈希
public class ConsistentHash implements LoadBalanceStrategy {
private final TreeMap<Long, Node> ring;
@Override
public Node select(List<Node> nodes, String key) {
Long hash = hash(key);
SortedMap<Long, Node> tailMap = ring.tailMap(hash);
return tailMap.isEmpty()
? ring.firstEntry().getValue()
: tailMap.firstEntry().getValue();
}
}
// 加权随机
public class WeightedRandom implements LoadBalanceStrategy {
@Override
public Node select(List<Node> nodes, String key) {
int totalWeight = nodes.stream().mapToInt(Node::getWeight).sum();
int random = ThreadLocalRandom.current().nextInt(totalWeight);
for (Node node : nodes) {
random -= node.getWeight();
if (random < 0) {
return node;
}
}
return nodes.get(0);
}
}
}
实际业务组合示例
1 电商库存系统
public class InventorySystem {
// 组合多种模式
private final ShardManager shardManager; // 数据分片
private final CacheManager cacheManager; // 多级缓存
private final EventProcessor eventProcessor; // 事件驱动
@Transactional
public boolean deductStock(String productId, int quantity) {
// 1. 获取分片
Shard shard = shardManager.getShard(productId);
// 2. 检查缓存
Integer cachedStock = cacheManager.getStock(productId);
// 3. 使用原子操作扣减
String lockKey = "stock:lock:" + productId;
if (tryLock(lockKey)) {
try {
// 4. 数据库操作
boolean success = shard.deductStock(productId, quantity);
// 5. 异步更新缓存和索引
CompletableFuture.runAsync(() -> {
cacheManager.updateStock(productId);
eventProcessor.publish(new StockChangeEvent(productId, quantity));
});
return success;
} finally {
unlock(lockKey);
}
}
return false;
}
}
2 社交推荐系统
public class RecommendationSystem {
// 组合实时和离线计算
private final RealtimeProcessor realtimeProcessor;
private final BatchProcessor batchProcessor;
private final HybridModel hybridModel;
public List<Recommendation> getRecommendations(String userId) {
// 1. 实时推荐（基于当前行为）
List<Recommendation> realtime = realtimeProcessor.process(userId);
// 2. 离线推荐（基于历史数据）
List<Recommendation> offline = batchProcessor.process(userId);
// 3. 混合模型组合
return hybridModel.combine(realtime, offline, new Combiner() {
@Override
public List<Recommendation> combine(
List<Recommendation> a,
List<Recommendation> b) {
// 加权组合
Map<String, Double> scores = new HashMap<>();
// 实时推荐权重高
for (Recommendation rec : a) {
scores.merge(rec.getItemId(),
rec.getScore() * 0.7, Double::sum);
}
// 离线推荐权重低
for (Recommendation rec : b) {
scores.merge(rec.getItemId(),
rec.getScore() * 0.3, Double::sum);
}
// 排序返回TopN
return scores.entrySet().stream()
.sorted(Map.Entry.<String, Double>comparingByValue().reversed())
.limit(50)
.map(entry -> new Recommendation(entry.getKey(), entry.getValue()))
.collect(Collectors.toList());
}
});
}
}
最佳实践建议：
-
数据分片选择
- 小数据量：范围分片
- 大数据量：一致性哈希
-
缓存策略
- 热点数据：多级缓存
- 冷数据：只查数据库
-
一致性方案
- 强一致性：分布式事务
- 最终一致性：事件驱动
-
性能优化
- 读写分离
- 异步处理
- 批量操作
-
监控告警
- 数据一致性检测
- 延迟监控
- 异常报警
根据业务场景选择合适的组合方式,平衡一致性、可用性和性能。
