本文目录导读：
我来系统性地介绍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());
}
});
}
}
最佳实践建议：
-
数据分片选择
- 小数据量：范围分片
- 大数据量：一致性哈希
-
缓存策略
- 热点数据：多级缓存
- 冷数据：只查数据库
-
一致性方案
- 强一致性：分布式事务
- 最终一致性：事件驱动
-
性能优化
- 读写分离
- 异步处理
- 批量操作
-
监控告警
- 数据一致性检测
- 延迟监控
- 异常报警
根据业务场景选择合适的组合方式,平衡一致性、可用性和性能。
版权声明：足球直播
足球直播
足球直播
足球直播
足球直播
中超直播
中超直播
中超直播
中超直播
中超直播
英超直播
英超直播
英超直播
英超直播
英超直播
德甲直播
德甲直播
德甲直播
德甲直播
德甲直播
西甲直播
西甲直播
西甲直播
西甲直播
西甲直播
意甲直播
意甲直播
意甲直播
意甲直播
意甲直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
篮球直播
篮球直播
篮球直播
篮球直播
篮球直播
足球直播
足球直播
足球直播
足球直播
足球直播
中超直播
中超直播
中超直播
中超直播
中超直播
英超直播
英超直播
英超直播
英超直播
英超直播
德甲直播
德甲直播
德甲直播
德甲直播
德甲直播
西甲直播
西甲直播
西甲直播
西甲直播
西甲直播
意甲直播
意甲直播
意甲直播
意甲直播
意甲直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
篮球直播
篮球直播
篮球直播
篮球直播
篮球直播
足球直播
足球直播
足球直播
足球直播
足球直播
中超直播
中超直播
中超直播
中超直播
中超直播
英超直播
英超直播
英超直播
英超直播
英超直播
德甲直播
德甲直播
德甲直播
德甲直播
德甲直播
西甲直播
西甲直播
西甲直播
西甲直播
西甲直播
意甲直播
意甲直播
意甲直播
意甲直播
意甲直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
篮球直播
篮球直播
篮球直播
篮球直播
篮球直播
