基于Java的自研分布式小文件存储系统设计源码

package com.ruyuan.dfs.namenode.server; import com.google.protobuf.InvalidProtocolBufferException; import com.ruyuan.dfs.common.Constants; import com.ruyuan.dfs.common.FileInfo; import com.ruyuan.dfs.common.NettyPacket; import com.ruyuan.dfs.common.enums.CommandType; import com.ruyuan.dfs.common.enums.NameNodeLaunchMode; import com.ruyuan.dfs.common.enums.NodeType; import com.ruyuan.dfs.common.enums.PacketType; import com.ruyuan.dfs.common.exception.NameNodeException; import com.ruyuan.dfs.common.exception.RequestTimeoutException; import com.ruyuan.dfs.common.metrics.Prometheus; import com.ruyuan.dfs.common.network.AbstractChannelHandler; import com.ruyuan.dfs.common.network.RequestWrapper; import com.ruyuan.dfs.common.network.file.FilePacket; import com.ruyuan.dfs.common.network.file.FileReceiveHandler; import com.ruyuan.dfs.common.utils.DefaultScheduler; import com.ruyuan.dfs.common.utils.NetUtils; import com.ruyuan.dfs.common.utils.PrettyCodes; import com.ruyuan.dfs.model.backup.*; import com.ruyuan.dfs.model.client.*; import com.ruyuan.dfs.model.common.DataNode; import com.ruyuan.dfs.model.datanode.*; import com.ruyuan.dfs.model.namenode.*; import com.ruyuan.dfs.namenode.config.NameNodeConfig; import com.ruyuan.dfs.namenode.datanode.DataNodeInfo; import com.ruyuan.dfs.namenode.datanode.DataNodeManager; import com.ruyuan.dfs.namenode.editslog.EditLogWrapper; import com.ruyuan.dfs.namenode.fs.CalculateResult; import com.ruyuan.dfs.namenode.fs.DiskNameSystem; import com.ruyuan.dfs.namenode.fs.Node; import com.ruyuan.dfs.namenode.rebalance.RemoveReplicaTask; import com.ruyuan.dfs.namenode.rebalance.ReplicaTask; import com.ruyuan.dfs.namenode.server.tomcat.domain.User; import com.ruyuan.dfs.namenode.shard.ShardingManager; import com.ruyuan.dfs.namenode.shard.controller.ControllerManager; import com.ruyuan.dfs.namenode.shard.peer.PeerNameNode; import com.ruyuan.dfs.namenode.shard.peer.PeerNameNodes; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.socket.SocketChannel; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.time.StopWatch; import java.io.File; import java.io.IOException; import java.util.*; import java.util.concurrent.Executor; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; /** * NameNode网络请求接口处理器 * * @author Sun Dasheng */ @Slf4j public class NameNodeApis extends AbstractChannelHandler { private final DefaultScheduler defaultScheduler; private final ControllerManager controllerManager; private final UserManager userManager; private final DiskNameSystem diskNameSystem; private final ShardingManager shardingManager; private final NameNodeConfig nameNodeConfig; private final PeerNameNodes peerNameNodes; private final DataNodeManager dataNodeManager; private final ThreadPoolExecutor executor; protected int nodeId; private final NameNodeLaunchMode mode; private final FetchEditLogBuffer fetchEditLogBuffer; private final AtomicBoolean slotsChanged = new AtomicBoolean(false); private Map<Integer, Integer> slots; private final FileReceiveHandler fileReceiveHandler; private BackupNodeInfoHolder backupNodeInfoHolder; public NameNodeApis(NameNodeConfig nameNodeConfig, DataNodeManager dataNodeManager, PeerNameNodes peerNameNodes, ShardingManager shardingManager, DiskNameSystem diskNameSystem, DefaultScheduler defaultScheduler, UserManager userManager, ControllerManager controllerManager) { this.dataNodeManager = dataNodeManager; this.peerNameNodes = peerNameNodes; this.nameNodeConfig = nameNodeConfig; this.shardingManager = shardingManager; this.diskNameSystem = diskNameSystem; this.nodeId = nameNodeConfig.getNameNodeId(); this.mode = nameNodeConfig.getMode(); this.userManager = userManager; this.peerNameNodes.setNameNodeApis(this); this.controllerManager = controllerManager; this.defaultScheduler = defaultScheduler; this.fetchEditLogBuffer = new FetchEditLogBuffer(diskNameSystem); this.shardingManager.addOnSlotAllocateCompletedListener(slots -> { if (slotsChanged.compareAndSet(false, true)) { this.slots = slots; } }); FsImageFileTransportCallback fsImageFileTransportCallback = new FsImageFileTransportCallback(nameNodeConfig, defaultScheduler, diskNameSystem); this.fileReceiveHandler = new FileReceiveHandler(fsImageFileTransportCallback); this.executor = new ThreadPoolExecutor(nameNodeConfig.getNameNodeApiCoreSize(), nameNodeConfig.getNameNodeApiMaximumPoolSize(), 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(nameNodeConfig.getNameNodeApiQueueSize())); } public BackupNodeInfoHolder getBackupNodeInfoHolder() { return backupNodeInfoHolder; } @Override protected Set<Integer> interestPackageTypes() { return new HashSet<>(); } @Override protected Executor getExecutor() { return executor; } @Override public void channelInactive(ChannelHandlerContext ctx) { String token = userManager.getTokenByChannel(ctx.channel()); String username = userManager.logout(ctx.channel()); if (username != null && token != null) { NettyPacket nettyPacket = NettyPacket.buildPacket(new byte[0], PacketType.CLIENT_LOGOUT); nettyPacket.setUsername(username); nettyPacket.setUserToken(token); broadcast(nettyPacket); } if (backupNodeInfoHolder != null && backupNodeInfoHolder.match(ctx.channel())) { backupNodeInfoHolder = null; } ctx.fireChannelInactive(); } @Override protected boolean handlePackage(ChannelHandlerContext ctx, NettyPacket request) { boolean consumedMsg = peerNameNodes.onMessage(request); if (consumedMsg) { return true; } if (request.isError()) { // 在请求转发的情况下，如果目标NameNode节点发生未知异常，然后返回的结果是异常的， // 而源NameNode正常处理流程会出现空指针异常，从而出现死循环。 log.warn("收到一个异常请求或响应, 丢弃不进行处理：[request={}]", request.getHeader()); return true; } PacketType packetType = PacketType.getEnum(request.getPacketType()); Prometheus.incCounter("namenode_net_package_inbound_count", "NameNode收到的请求数量"); Prometheus.incCounter("namenode_net_package_inbound_bytes", "NameNode收到的请求大小", request.getBody().length); StopWatch stopWatch = new StopWatch(); stopWatch.start(); RequestWrapper requestWrapper = new RequestWrapper(ctx, request, nodeId, bodyLength -> { Prometheus.incCounter("namenode_net_package_outbound_count", "NameNode返回的响应数量"); Prometheus.incCounter("namenode_net_package_outbound_bytes", "NameNode返回的响应大小", bodyLength); stopWatch.stop(); long time = stopWatch.getTime(); Prometheus.gauge("namenode_request_times", "NameNode处理请求耗时", "requestType", packetType.getDescription(), time); }); try { switch (packetType) { case DATA_NODE_REGISTER: handleDataNodeRegisterRequest(requestWrapper); break; case HEART_BRET: handleDataNodeHeartbeatRequest(requestWrapper); break; case MKDIR: handleMkdirRequest(requestWrapper); break; case FETCH_EDIT_LOG: