临时节点注册过程

开始测试源码时，发现注册接口的断点无论如何都不会执行，打印语句也不会执行，最后发现nacos设置了一个过滤器，会根据注册服务的ip端口经过hash后的值进行负载均衡，选择一个节点处理请求，所以并不一定会在请求的节点进入服务注册接口，下面就是过滤器doFilter代码，这个过滤器只会对/v1/ns/相关接口生效

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain)
throws IOException, ServletException {
    ReuseHttpServletRequest req = new ReuseHttpServletRequest((HttpServletRequest) servletRequest);
    HttpServletResponse resp = (HttpServletResponse) servletResponse;

    String urlString = req.getRequestURI();

    if (StringUtils.isNotBlank(req.getQueryString())) {
        urlString += "?" + req.getQueryString();
    }

    try {
        Method method = controllerMethodsCache.getMethod(req);

        String path = new URI(req.getRequestURI()).getPath();
        if (method == null) {
            throw new NoSuchMethodException(req.getMethod() + " " + path);
        }

        if (!method.isAnnotationPresent(CanDistro.class)) {
            filterChain.doFilter(req, resp);
            return;
        }
        String distroTag = distroTagGenerator.getResponsibleTag(req);

        if (distroMapper.responsible(distroTag)) {
            filterChain.doFilter(req, resp);
            return;
        }

        // proxy request to other server if necessary:
        String userAgent = req.getHeader(HttpHeaderConsts.USER_AGENT_HEADER);

        if (StringUtils.isNotBlank(userAgent) && userAgent.contains(UtilsAndCommons.NACOS_SERVER_HEADER)) {
            // This request is sent from peer server, should not be redirected again:
            Loggers.SRV_LOG.error("receive invalid redirect request from peer {}", req.getRemoteAddr());
            resp.sendError(HttpServletResponse.SC_BAD_REQUEST,
                           "receive invalid redirect request from peer " + req.getRemoteAddr());
            return;
        }

        final String targetServer = distroMapper.mapSrv(distroTag);

        List<String> headerList = new ArrayList<>(16);
        Enumeration<String> headers = req.getHeaderNames();
        while (headers.hasMoreElements()) {
            String headerName = headers.nextElement();
            headerList.add(headerName);
            headerList.add(req.getHeader(headerName));
        }

        final String body = IoUtils.toString(req.getInputStream(), StandardCharsets.UTF_8.name());
        final Map<String, String> paramsValue = HttpClient.translateParameterMap(req.getParameterMap());

        RestResult<String> result = HttpClient
        .request(HTTP_PREFIX + targetServer + req.getRequestURI(), headerList, paramsValue, body,
                 PROXY_CONNECT_TIMEOUT, PROXY_READ_TIMEOUT, StandardCharsets.UTF_8.name(), req.getMethod());
        String data = result.ok() ? result.getData() : result.getMessage();
        try {
            WebUtils.response(resp, data, result.getCode());
        } catch (Exception ignore) {
            Loggers.SRV_LOG.warn("[DISTRO-FILTER] request failed: " + distroMapper.mapSrv(distroTag) + urlString);
        }
    } catch (AccessControlException e) {
        resp.sendError(HttpServletResponse.SC_FORBIDDEN, "access denied: " + ExceptionUtil.getAllExceptionMsg(e));
    } catch (NoSuchMethodException e) {
        resp.sendError(HttpServletResponse.SC_NOT_IMPLEMENTED,
        "no such api:" + req.getMethod() + ":" + req.getRequestURI());
    } catch (Exception e) {
        Loggers.SRV_LOG.warn("[DISTRO-FILTER] Server failed: ", e);
        resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
        "Server failed, " + ExceptionUtil.getAllExceptionMsg(e));
    }

    }

经过过滤器后，会在对应的节点进入服务注册的接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@CanDistro
@PostMapping
@Secured(action = ActionTypes.WRITE)
public String register(HttpServletRequest request) throws Exception {
    final String namespaceId = WebUtils
    .optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
    final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
    NamingUtils.checkServiceNameFormat(serviceName);

    final Instance instance = HttpRequestInstanceBuilder.newBuilder()
    .setDefaultInstanceEphemeral(switchDomain.isDefaultInstanceEphemeral()).setRequest(request).build();

    getInstanceOperator().registerInstance(namespaceId, serviceName, instance);
    NotifyCenter.publishEvent(new RegisterInstanceTraceEvent(System.currentTimeMillis(), "", false, namespaceId,
                                                             NamingUtils.getGroupName(serviceName), NamingUtils.getServiceName(serviceName), instance.getIp(),
                                                             instance.getPort()));
    return "ok";
}

经过不断调用，会进入到ClientOperationServiceProxy类

1
2
3
4
5
@Override
public void registerInstance(Service service, Instance instance, String clientId) throws NacosException {
    final ClientOperationService operationService = chooseClientOperationService(instance);
    operationService.registerInstance(service, instance, clientId);
}

这里使用了策略模式，根据服务是否是临时节点，选择响应的处理方法，这里我们先讲临时节点注册过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Override
public void registerInstance(Service service, Instance instance, String clientId) throws NacosException {
    NamingUtils.checkInstanceIsLegal(instance);

    Service singleton = ServiceManager.getInstance().getSingleton(service);
    if (!singleton.isEphemeral()) {
        throw new NacosRuntimeException(NacosException.INVALID_PARAM,
                                        String.format("Current service %s is persistent service, can't register ephemeral instance.",
                                                      singleton.getGroupedServiceName()));
    }
    Client client = clientManager.getClient(clientId);
    if (!clientIsLegal(client, clientId)) {
        return;
    }
    InstancePublishInfo instanceInfo = getPublishInfo(instance);
    client.addServiceInstance(singleton, instanceInfo);
    client.setLastUpdatedTime();
    client.recalculateRevision();
    NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId));
    NotifyCenter
    .publishEvent(new MetadataEvent.InstanceMetadataEvent(singleton, instanceInfo.getMetadataId(), false));
}

这里会判断一下要注册的服务所在的服务组是否也是临时的，不是就会报错，这个方法中比较重要的是client.addServiceInstance(singleton, instanceInfo)，负责添加服务实例，并且推送给其他节点

1
2
3
4
5
6
7
8
9
10
11
12
public boolean addServiceInstance(Service service, InstancePublishInfo instancePublishInfo) {
    if (null == publishers.put(service, instancePublishInfo)) {
        if (instancePublishInfo instanceof BatchInstancePublishInfo) {
            MetricsMonitor.incrementIpCountWithBatchRegister(instancePublishInfo);
        } else {
            MetricsMonitor.incrementInstanceCount();
        }
    }
    NotifyCenter.publishEvent(new ClientEvent.ClientChangedEvent(this));
    Loggers.SRV_LOG.info("Client change for service {}, {}", service, getClientId());
    return true;
}

NotifyCenter.publishEvent(new ClientEvent.ClientChangedEvent(this))将服务节点改变推送给其他nacos节点，每个事件都会有对应的发布者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
    if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
        return INSTANCE.sharePublisher.publish(event);
    }

    final String topic = ClassUtils.getCanonicalName(eventType);

    EventPublisher publisher = INSTANCE.publisherMap.get(topic);
    if (publisher != null) {
        return publisher.publish(event);
    }
    if (event.isPluginEvent()) {
        return true;
    }
    LOGGER.warn("There are no [{}] publishers for this event, please register", topic);
    return false;
}

这里会根据事件类型获取对应的发布者，发布者会有响应的订阅者（subscribers），像服务节点改变事件的订阅者就是所有其他nacos节点，这里用到了设计模式中的观察者模式，订阅者就是观察者，发布者具体发布的代码可参考com.alibaba.nacos.naming.core.v2.event.publisher.NamingEventPublisherr类

1
2
3
4
5
6
7
8
9
public boolean publish(Event event) {
        checkIsStart();
        boolean success = this.queue.offer(event);
        if (!success) {
            Loggers.EVT_LOG.warn("Unable to plug in due to interruption, synchronize sending time, event : {}", event);
            handleEvent(event);
        }
        return true;
    }

原理就是发布者会将事件存放到BlockingQueue中，另外一个方法不断的从BlockingQueue中获取并执行事件，通知订阅者

1
2
3
4
5
6
7
8
9
10
11
12
private void handleEvents() {
    while (!shutdown) {
        try {
            final Event event = queue.take();
            handleEvent(event);
        } catch (InterruptedException e) {
            Loggers.EVT_LOG.warn("Naming Event Publisher {} take event from queue failed:", this.publisherName, e);
            // set the interrupted flag
            Thread.currentThread().interrupt();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
private void handleEvent(Event event) {
    Class<? extends Event> eventType = event.getClass();
    Set<Subscriber<? extends Event>> subscribers = subscribes.get(eventType);
    if (null == subscribers) {
        if (Loggers.EVT_LOG.isDebugEnabled()) {
            Loggers.EVT_LOG.debug("[NotifyCenter] No subscribers for slow event {}", eventType.getName());
        }
        return;
    }
    for (Subscriber subscriber : subscribers) {
        notifySubscriber(subscriber, event);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public void notifySubscriber(Subscriber subscriber, Event event) {
    if (Loggers.EVT_LOG.isDebugEnabled()) {
        Loggers.EVT_LOG.debug("[NotifyCenter] the {} will received by {}", event, subscriber);
    }
    final Runnable job = () -> subscriber.onEvent(event);
    final Executor executor = subscriber.executor();
    if (executor != null) {
        executor.execute(job);
    } else {
        try {
            job.run();
        } catch (Throwable e) {
            Loggers.EVT_LOG.error("Event callback exception: ", e);
        }
    }
}

这几个方法就是将事件推送给所有订阅者，subscriber.onEvent(event)就是将事件提交给订阅者，后面可能还会涉及到很多阻塞队列的使用，阻塞队列可以很好的应对并发问题，具体可以参考源码，推送代码最后面会会构造一个_type属性为DistroDataRequest的com.alibaba.nacos.api.grpc.auto.Payload，并调用io.grpc.ClientCall#sendMessage方法，提交给其他nacos节点，这是grpc客户端的接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
public abstract class ClientCall<ReqT, RespT> {
    public ClientCall() {
    }

    public abstract void start(Listener<RespT> var1, Metadata var2);

    public abstract void request(int var1);

    public abstract void cancel(@Nullable String var1, @Nullable Throwable var2);

    public abstract void halfClose();

    public abstract void sendMessage(ReqT var1);

    public boolean isReady() {
        return true;
    }

    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1703")
    public void setMessageCompression(boolean enabled) {
    }

    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/2607")
    public Attributes getAttributes() {
        return Attributes.EMPTY;
    }

    public abstract static class Listener<T> {
        public Listener() {
        }

        public void onHeaders(Metadata headers) {
        }

        public void onMessage(T message) {
        }

        public void onClose(Status status, Metadata trailers) {
        }

        public void onReady() {
        }
    }
}

这个接口中通过sendMessage推送消息，以及通过onMessage异步接收响应，主要作用就是发起gRpc请求以及异步接收服务端响应

临时节点nacos响应过程

上面提到非临时节点注册过程中会通过ClientCall接口将注册事件提交给其他nacos节点，其他nacos节点则通过ServerCall接收来自ClientCall的消息，ServerCall是服务端的接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
public abstract class ServerCall<ReqT, RespT> {
    public ServerCall() {
    }

    public abstract void request(int var1);

    public abstract void sendHeaders(Metadata var1);

    public abstract void sendMessage(RespT var1);

    public boolean isReady() {
        return true;
    }

    public abstract void close(Status var1, Metadata var2);

    public abstract boolean isCancelled();

    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1704")
    public void setMessageCompression(boolean enabled) {
    }

    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1704")
    public void setCompression(String compressor) {
    }

    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/4692")
    public SecurityLevel getSecurityLevel() {
        return SecurityLevel.NONE;
    }

    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1779")
    public Attributes getAttributes() {
        return Attributes.EMPTY;
    }

    @Nullable
    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/2924")
    public String getAuthority() {
        return null;
    }

    public abstract MethodDescriptor<ReqT, RespT> getMethodDescriptor();

    public abstract static class Listener<ReqT> {
        public Listener() {
        }

        public void onMessage(ReqT message) {
        }

        public void onHalfClose() {
        }

        public void onCancel() {
        }

        public void onComplete() {
        }

        public void onReady() {
        }
    }
}

ServerCall与ClientCall相反，通过onMessage接收来自ClientCall的消息，并且用sendMessage对ClientCall异步提交响应，具体处理入口是onHalfClose方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public void onHalfClose() {
    if (this.canInvoke) {
        if (this.request == null) {
            this.call.close(Status.INTERNAL.withDescription("Half-closed without a request"), new Metadata());
        } else {
            UnaryServerCallHandler.this.method.invoke(this.request, this.responseObserver);
            this.request = null;
            this.responseObserver.freeze();
            if (this.wasReady) {
                this.onReady();
            }

        }
    }
}

其中UnaryServerCallHandler.this.method.invoke(this.request, this.responseObserver);就是具体处理方法的入口，这是因为注册了相关Payload消息的事件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
private void addServices(MutableHandlerRegistry handlerRegistry, ServerInterceptor... serverInterceptor) {

    // unary common call register.
    final MethodDescriptor<Payload, Payload> unaryPayloadMethod = MethodDescriptor.<Payload, Payload>newBuilder()
    .setType(MethodDescriptor.MethodType.UNARY)
    .setFullMethodName(MethodDescriptor.generateFullMethodName(GrpcServerConstants.REQUEST_SERVICE_NAME,
                                                               GrpcServerConstants.REQUEST_METHOD_NAME))
    .setRequestMarshaller(ProtoUtils.marshaller(Payload.getDefaultInstance()))
    .setResponseMarshaller(ProtoUtils.marshaller(Payload.getDefaultInstance())).build();

    final ServerCallHandler<Payload, Payload> payloadHandler = ServerCalls
    .asyncUnaryCall((request, responseObserver) -> grpcCommonRequestAcceptor.request(request, responseObserver));

    final ServerServiceDefinition serviceDefOfUnaryPayload = ServerServiceDefinition.builder(
        GrpcServerConstants.REQUEST_SERVICE_NAME)
    .addMethod(unaryPayloadMethod, payloadHandler).build();
    handlerRegistry.addService(ServerInterceptors.intercept(serviceDefOfUnaryPayload, serverInterceptor));

    // bi stream register.
    final ServerCallHandler<Payload, Payload> biStreamHandler = ServerCalls.asyncBidiStreamingCall(
        (responseObserver) -> grpcBiStreamRequestAcceptor.requestBiStream(responseObserver));

    final MethodDescriptor<Payload, Payload> biStreamMethod = MethodDescriptor.<Payload, Payload>newBuilder()
    .setType(MethodDescriptor.MethodType.BIDI_STREAMING).setFullMethodName(MethodDescriptor
                                                                           .generateFullMethodName(GrpcServerConstants.REQUEST_BI_STREAM_SERVICE_NAME,
                                                                                                   GrpcServerConstants.REQUEST_BI_STREAM_METHOD_NAME))
    .setRequestMarshaller(ProtoUtils.marshaller(Payload.newBuilder().build()))
    .setResponseMarshaller(ProtoUtils.marshaller(Payload.getDefaultInstance())).build();

    final ServerServiceDefinition serviceDefOfBiStream = ServerServiceDefinition
    .builder(GrpcServerConstants.REQUEST_BI_STREAM_SERVICE_NAME).addMethod(biStreamMethod, biStreamHandler).build();
    handlerRegistry.addService(ServerInterceptors.intercept(serviceDefOfBiStream, serverInterceptor));

}

该方法中注册了相关事件，其中11到17行就是注册相关Payload消息的事件，也就是我们要说的事件，我们可以看到注册过程中，用到了grpcCommonRequestAcceptor.request方法，即com.alibaba.nacos.core.remote.grpc.GrpcRequestAcceptor#request方法，这个方法会根据Payload的类型（即PayLoad的_type属性）调用响应的处理方法，我们这里说的临时节点通知消息就是DistroDataRequestHandler类的handle方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public DistroDataResponse handle(DistroDataRequest request, RequestMeta meta) throws NacosException {
    try {
        switch (request.getDataOperation()) {
            case VERIFY:
                return handleVerify(request.getDistroData(), meta);
            case SNAPSHOT:
                return handleSnapshot();
            case ADD:
            case CHANGE:
            case DELETE:
                return handleSyncData(request.getDistroData());
            case QUERY:
                return handleQueryData(request.getDistroData());
            default:
                return new DistroDataResponse();
        }
    } catch (Exception e) {
        Loggers.DISTRO.error("[DISTRO-FAILED] distro handle with exception", e);
        DistroDataResponse result = new DistroDataResponse();
        result.setErrorCode(ResponseCode.FAIL.getCode());
        result.setMessage("handle distro request with exception");
        return result;
    }
}

该方法会将其他nacos节点推送过来的消息应用到本节点

非临时节点注册过程

在上面ClientOperationServiceProxy中提到使用了策略模式，下面是非临时节点处理方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@Override
public void registerInstance(Service service, Instance instance, String clientId) {
    Service singleton = ServiceManager.getInstance().getSingleton(service);
    if (singleton.isEphemeral()) {
        throw new NacosRuntimeException(NacosException.INVALID_PARAM,
                                        String.format("Current service %s is ephemeral service, can't register persistent instance.",
                                                      singleton.getGroupedServiceName()));
    }
    final InstanceStoreRequest request = new InstanceStoreRequest();
    request.setService(service);
    request.setInstance(instance);
    request.setClientId(clientId);
    final WriteRequest writeRequest = WriteRequest.newBuilder().setGroup(group())
    .setData(ByteString.copyFrom(serializer.serialize(request))).setOperation(DataOperation.ADD.name())
    .build();

    try {
        protocol.write(writeRequest);
        Loggers.RAFT.info("Client registered. service={}, clientId={}, instance={}", service, instance, clientId);
    } catch (Exception e) {
        throw new NacosRuntimeException(NacosException.SERVER_ERROR, e);
    }
}

该方法中需要注意的就是protocol.write(writeRequest)，从这里可以看出来，非临时节点和临时的节点的区别之一就是非临时节点会用到jraft算法，而临时节点会直接存储然后广播

1
2
3
4
5
6
@Override
public Response write(WriteRequest request) throws Exception {
    CompletableFuture<Response> future = writeAsync(request);
    // Here you wait for 10 seconds, as long as possible, for the request to complete
    return future.get(10_000L, TimeUnit.MILLISECONDS);
}

这里异步调用异步请求方法，nacos源码中涉及到很多的异步方法以及异步请求，所以源码观看起来可能有点麻烦，writeAsync(request)调用的是JRaftServer的commit方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public CompletableFuture<Response> commit(final String group, final Message data,
                                          final CompletableFuture<Response> future) {
    LoggerUtils.printIfDebugEnabled(Loggers.RAFT, "data requested this time : {}", data);
    final RaftGroupTuple tuple = findTupleByGroup(group);
    if (tuple == null) {
        future.completeExceptionally(new IllegalArgumentException("No corresponding Raft Group found : " + group));
        return future;
    }

    FailoverClosureImpl closure = new FailoverClosureImpl(future);

    final Node node = tuple.node;
    if (node.isLeader()) {
        // The leader node directly applies this request
        applyOperation(node, data, closure);
    } else {
        // Forward to Leader for request processing
        invokeToLeader(group, data, 10000, closure);
    }
    return future;
}

这里可以看出来jraft算法的特点，请求需要交给leader及主节点处理，如果是主节点就直接调用applyOperation(node, data, closure)处理，如果不是主节点，就会调用invokeToLeader将请求打包成WriteRequest并请求主节点，主节点的响应方法是

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
protected void handleRequest(final JRaftServer server, final String group, final RpcContext rpcCtx, Message message) {
    try {
        final JRaftServer.RaftGroupTuple tuple = server.findTupleByGroup(group);
        if (Objects.isNull(tuple)) {
            rpcCtx.sendResponse(Response.newBuilder().setSuccess(false)
                                .setErrMsg("Could not find the corresponding Raft Group : " + group).build());
            return;
        }
        if (tuple.getNode().isLeader()) {
            execute(server, rpcCtx, message, tuple);
        } else {
            rpcCtx.sendResponse(
                Response.newBuilder().setSuccess(false).setErrMsg("Could not find leader : " + group).build());
        }
    } catch (Throwable e) {
        Loggers.RAFT.error("handleRequest has error : ", e);
        rpcCtx.sendResponse(Response.newBuilder().setSuccess(false).setErrMsg(e.toString()).build());
    }
}

其中execute(server, rpcCtx, message, tuple)会调用上述commit中也提到过的applyOperation(node, data, closure)方法，applyOperation方法后续过程用到了LMAX Disruptor框架，观看源码时需要注意，applyOperation会调用到com.alipay.sofa.jraft.storage.impl.LogManagerImpl#appendEntries，后面将非临时节点存储到本地日志，并通过io.grpc.ClientCall将事件和日志打包成com.alipay.sofa.jraft.rpc.RpcRequests.AppendEntriesRequest并通知到其他节点，通知其他节点时，会继续异步调用，后续过程中会调用到以下方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@Override
public Response onApply(WriteRequest request) {
final Lock lock = readLock;
lock.lock();
try {
    final InstanceStoreRequest instanceRequest = serializer.deserialize(request.getData().toByteArray());
    final DataOperation operation = DataOperation.valueOf(request.getOperation());
    switch (operation) {
        case ADD:
            onInstanceRegister(instanceRequest.service, instanceRequest.instance,
                               instanceRequest.getClientId());
            break;
        case DELETE:
            onInstanceDeregister(instanceRequest.service, instanceRequest.getClientId());
            break;
        case CHANGE:
            if (instanceAndServiceExist(instanceRequest)) {
                onInstanceRegister(instanceRequest.service, instanceRequest.instance,
                                   instanceRequest.getClientId());
            }
            break;
        default:
            return Response.newBuilder().setSuccess(false).setErrMsg("unsupport operation : " + operation)
            .build();
    }
    return Response.newBuilder().setSuccess(true).build();
} catch (Exception e) {
    Loggers.RAFT.warn("Persistent client operation failed. ", e);
    return Response.newBuilder().setSuccess(false)
    .setErrMsg("Persistent client operation failed. " + e.getMessage()).build();
} finally {
    lock.unlock();
}
}

这个方法主要就是注册或者删除非临时节点的后续处理

非临时节点nacos响应过程

上面提到，applyOperation方法会将事件打包后通过ClientCall发送给其他节点，参考上面临时节点nacos响应过程，可知，我们只需要找到ServerCall的onHalfClose方法中，AppendEntriesRequest请求对应的处理类即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
public void registerProcessor(RpcProcessor processor) {
String interest = processor.interest();
Message reqIns = (Message)Requires.requireNonNull(this.parserClasses.get(interest), "null default instance: " + interest);
MethodDescriptor<Message, Message> method = MethodDescriptor.newBuilder().setType(MethodType.UNARY).setFullMethodName(MethodDescriptor.generateFullMethodName(processor.interest(), "_call")).setRequestMarshaller(ProtoUtils.marshaller(reqIns)).setResponseMarshaller(ProtoUtils.marshaller(this.marshallerRegistry.findResponseInstanceByRequest(interest))).build();
ServerCallHandler<Message, Message> handler = ServerCalls.asyncUnaryCall((request, responseObserver) -> {
    final SocketAddress remoteAddress = RemoteAddressInterceptor.getRemoteAddress();
    final Connection conn = ConnectionInterceptor.getCurrentConnection(this.closedEventListeners);
    RpcContext rpcCtx = new RpcContext() {
        public void sendResponse(Object responseObj) {
            try {
                responseObserver.onNext((Message)responseObj);
                responseObserver.onCompleted();
            } catch (Throwable var3) {
                GrpcServer.LOG.warn("[GRPC] failed to send response.", var3);
            }

        }

        public Connection getConnection() {
            if (conn == null) {
                throw new IllegalStateException("fail to get connection");
            } else {
                return conn;
            }
        }

        public String getRemoteAddress() {
            return remoteAddress != null ? remoteAddress.toString() : null;
        }
    };
    RpcProcessor.ExecutorSelector selector = processor.executorSelector();
    Object executor;
    if (selector != null && request instanceof RpcRequests.AppendEntriesRequest) {
        RpcRequests.AppendEntriesRequest req = (RpcRequests.AppendEntriesRequest)request;
        RpcRequests.AppendEntriesRequestHeader.Builder header = AppendEntriesRequestHeader.newBuilder().setGroupId(req.getGroupId()).setPeerId(req.getPeerId()).setServerId(req.getServerId());
        executor = selector.select(interest, header.build());
    } else {
        executor = processor.executor();
    }

    if (executor == null) {
        executor = this.defaultExecutor;
    }

    if (executor != null) {
        ((Executor)executor).execute(() -> {
            processor.handleRequest(rpcCtx, request);
        });
    } else {
        processor.handleRequest(rpcCtx, request);
    }

});
ServerServiceDefinition serviceDef = ServerServiceDefinition.builder(interest).addMethod(method, handler).build();
this.handlerRegistry.addService(ServerInterceptors.intercept(serviceDef, (ServerInterceptor[])this.serverInterceptors.toArray(new ServerInterceptor[0])));
}

第一行的String interest = processor.interest();就是获取处理器对应的请求类型，后续进行注册，nacos中大部分请求都需要通过实现RpcProcessor处理，而registerProcessor方法就是注册每个请求对应的处理类，比如AppendEntriesRequest请求注册的就是AppendEntriesRequestProcessor，上面提到过的WriteRequest对应的就是NacosWriteRequestProcessor以及NacosReadRequestProcessor，这两个类都是AbstractProcessor的子类，所以实际上是在AbstractProcessor中处理的，AppendEntriesRequestProcessor实际调用的方法是processRequest0

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public Message processRequest0(RaftServerService service, RpcRequests.AppendEntriesRequest request, RpcRequestClosure done) {
    Node node = (Node)service;
    if (node.getRaftOptions().isReplicatorPipeline()) {
        String groupId = request.getGroupId();
        PeerPair pair = this.pairOf(request.getPeerId(), request.getServerId());
        boolean isHeartbeat = this.isHeartbeatRequest(request);
        int reqSequence = -1;
        if (!isHeartbeat) {
            reqSequence = this.getAndIncrementSequence(groupId, pair, done.getRpcCtx().getConnection());
        }

        Message response = service.handleAppendEntriesRequest(request, new SequenceRpcRequestClosure(done, this.defaultResp(), groupId, pair, reqSequence, isHeartbeat));
        if (response != null) {
            if (isHeartbeat) {
                done.getRpcCtx().sendResponse(response);
            } else {
                this.sendSequenceResponse(groupId, pair, reqSequence, done.getRpcCtx(), response);
            }
        }

        return null;
    } else {
        return service.handleAppendEntriesRequest(request, done);
    }
}

service.handleAppendEntriesRequest(request, done)就是将主节点推送过来的日志应用到本地，也会和主节点一样调用到com.alipay.sofa.jraft.storage.impl.LogManagerImpl#appendEntries

applyOperation具体调用过程

参考文章：蚂蚁金服生产级 Raft 算法库 SOFAJRaft 存储模块剖析 | SOFAJRaft 实现原理

这里就是讲一下，applyOperation具体调用过程，也就是nacos的jraft算法过程中，主节点处理消息的过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public void applyOperation(Node node, Message data, FailoverClosure closure) {
    final Task task = new Task();
    task.setDone(new NacosClosure(data, status -> {
        NacosClosure.NacosStatus nacosStatus = (NacosClosure.NacosStatus) status;
        closure.setThrowable(nacosStatus.getThrowable());
        closure.setResponse(nacosStatus.getResponse());
        closure.run(nacosStatus);
    }));

    // add request type field at the head of task data.
    byte[] requestTypeFieldBytes = new byte[2];
    requestTypeFieldBytes[0] = ProtoMessageUtil.REQUEST_TYPE_FIELD_TAG;
    if (data instanceof ReadRequest) {
        requestTypeFieldBytes[1] = ProtoMessageUtil.REQUEST_TYPE_READ;
    } else {
        requestTypeFieldBytes[1] = ProtoMessageUtil.REQUEST_TYPE_WRITE;
    }

    byte[] dataBytes = data.toByteArray();
    task.setData((ByteBuffer) ByteBuffer.allocate(requestTypeFieldBytes.length + dataBytes.length)
                  .put(requestTypeFieldBytes).put(dataBytes).position(0));
    node.apply(task);
}

可以看到applyOperation方法本身就是生成一个task，然后调用node.apply(task)，这里node是Node的实现类

：NodeImpl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public void apply(Task task) {
if (this.shutdownLatch != null) {
    ThreadPoolsFactory.runClosureInThread(this.groupId, task.getDone(), new Status(RaftError.ENODESHUTDOWN, "Node is shutting down.", new Object[0]));
    throw new IllegalStateException("Node is shutting down");
} else {
    Requires.requireNonNull(task, "Null task");
    LogEntry entry = new LogEntry();
    entry.setData(task.getData());
    EventTranslator<LogEntryAndClosure> translator = (event, sequence) -> {
        event.reset();
        event.done = task.getDone();
        event.entry = entry;
        event.expectedTerm = task.getExpectedTerm();
    };
    switch (this.options.getApplyTaskMode()) {
        case Blocking:
            this.applyQueue.publishEvent(translator);
            break;
        case NonBlocking:
        default:
            if (!this.applyQueue.tryPublishEvent(translator)) {
                String errorMsg = "Node is busy, has too many tasks, queue is full and bufferSize=" + this.applyQueue.getBufferSize();
                ThreadPoolsFactory.runClosureInThread(this.groupId, task.getDone(), new Status(RaftError.EBUSY, errorMsg, new Object[0]));
                LOG.warn("Node {} applyQueue is overload.", this.getNodeId());
                this.metrics.recordTimes("apply-task-overload-times", 1L);
                if (task.getDone() == null) {
                    throw new OverloadException(errorMsg);
                }
            }
    }

}
}

可以看到，apply方法就是生成一个translator，然后通过applyQueue发布，这里的applyQueue是RingBuffer的对象，RingBuffer就是上面提到的LMAX Disruptor框架的内容，而且这个类使用了装饰器模式，既然使用到了LMAX Disruptor框架，那么实现肯定有一个实现EventHandler的事件处理类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
private class LogEntryAndClosureHandler implements EventHandler<LogEntryAndClosure> {
    private final List<LogEntryAndClosure> tasks;

    private LogEntryAndClosureHandler() {
        this.tasks = new ArrayList(NodeImpl.this.raftOptions.getApplyBatch());
    }

    public void onEvent(LogEntryAndClosure event, long sequence, boolean endOfBatch) throws Exception {
        if (event.shutdownLatch != null) {
            if (!this.tasks.isEmpty()) {
                NodeImpl.this.executeApplyingTasks(this.tasks);
                this.reset();
            }

            int num = NodeImpl.GLOBAL_NUM_NODES.decrementAndGet();
            NodeImpl.LOG.info("The number of active nodes decrement to {}.", num);
            event.shutdownLatch.countDown();
        } else {
            this.tasks.add(event);
            if (this.tasks.size() >= NodeImpl.this.raftOptions.getApplyBatch() || endOfBatch) {
                NodeImpl.this.executeApplyingTasks(this.tasks);
                this.reset();
            }

        }
    }

    private void reset() {
        Iterator var1 = this.tasks.iterator();

        while(var1.hasNext()) {
            LogEntryAndClosure task = (LogEntryAndClosure)var1.next();
            task.reset();
        }

        this.tasks.clear();
    }
}

这里需要注意的是executeApplyingTasks方法，它会调用上面提到过的com.alipay.sofa.jraft.storage.impl.LogManagerImpl#appendEntries方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
public void appendEntries(List<LogEntry> entries, LogManager.StableClosure done) {
    assert done != null;

    Requires.requireNonNull(done, "done");
    if (this.hasError) {
        entries.clear();
        ThreadPoolsFactory.runClosureInThread(this.groupId, done, new Status(RaftError.EIO, "Corrupted LogStorage", new Object[0]));
    } else {
        boolean doUnlock = true;
        this.writeLock.lock();

        try {
            if (!entries.isEmpty() && !this.checkAndResolveConflict(entries, done, this.writeLock)) {
                entries.clear();
                return;
            }

            for(int i = 0; i < entries.size(); ++i) {
                LogEntry entry = (LogEntry)entries.get(i);
                if (this.raftOptions.isEnableLogEntryChecksum()) {
                    entry.setChecksum(entry.checksum());
                }

                if (entry.getType() == EntryType.ENTRY_TYPE_CONFIGURATION) {
                    Configuration oldConf = new Configuration();
                    if (entry.getOldPeers() != null) {
                        oldConf = new Configuration(entry.getOldPeers(), entry.getOldLearners());
                    }

                    ConfigurationEntry conf = new ConfigurationEntry(entry.getId(), new Configuration(entry.getPeers(), entry.getLearners()), oldConf);
                    this.configManager.add(conf);
                }
            }

            if (!entries.isEmpty()) {
                done.setFirstLogIndex(((LogEntry)entries.get(0)).getId().getIndex());
                this.logsInMemory.addAll(entries);
            }

            done.setEntries(entries);
            doUnlock = false;
            if (!this.wakeupAllWaiter(this.writeLock)) {
                this.notifyLastLogIndexListeners();
            }

            this.diskQueue.publishEvent((event, sequence) -> {
                event.reset();
                event.type = LogManagerImpl.EventType.OTHER;
                event.done = done;
            });
        } finally {
            if (doUnlock) {
                this.writeLock.unlock();
            }

        }

    }
}

这里需要注意的是this.wakeupAllWaiter(this.writeLock)，这个方法就是nacos主节点唤醒并通知其他nacos节点，它后面会调用到sendEntries方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
private boolean sendEntries(long nextSendingIndex) {
    RpcRequests.AppendEntriesRequest.Builder rb = AppendEntriesRequest.newBuilder();
    if (!this.fillCommonFields(rb, nextSendingIndex - 1L, false)) {
        this.installSnapshot();
        return false;
    } else {
        ByteBufferCollector dataBuf = null;
        int maxEntriesSize = this.raftOptions.getMaxEntriesSize();
        RecyclableByteBufferList byteBufList = RecyclableByteBufferList.newInstance();

        label152: {
            boolean var19;
            try {
                for(int i = 0; i < maxEntriesSize; ++i) {
                    RaftOutter.EntryMeta.Builder emb = EntryMeta.newBuilder();
                    if (!this.prepareEntry(nextSendingIndex, i, emb, byteBufList)) {
                        break;
                    }

                    rb.addEntries(emb.build());
                }

                if (rb.getEntriesCount() != 0) {
                    if (byteBufList.getCapacity() <= 0) {
                        break label152;
                    }

                    dataBuf = ByteBufferCollector.allocateByRecyclers(byteBufList.getCapacity());
                    Iterator var20 = byteBufList.iterator();

                    while(var20.hasNext()) {
                        ByteBuffer b = (ByteBuffer)var20.next();
                        dataBuf.put(b);
                    }

                    ByteBuffer buf = dataBuf.getBuffer();
                    BufferUtils.flip(buf);
                    rb.setData(ZeroByteStringHelper.wrap(buf));
                    break label152;
                }

                if (nextSendingIndex < this.options.getLogManager().getFirstLogIndex()) {
                    this.installSnapshot();
                    var19 = false;
                    return var19;
                }

                this.waitMoreEntries(nextSendingIndex);
                var19 = false;
            } finally {
                RecycleUtil.recycle(byteBufList);
            }

            return var19;
        }

        final RpcRequests.AppendEntriesRequest request = rb.build();
        if (LOG.isDebugEnabled()) {
            LOG.debug("Node {} send AppendEntriesRequest to {} term {} lastCommittedIndex {} prevLogIndex {} prevLogTerm {} logIndex {} count {}", new Object[]{this.options.getNode().getNodeId(), this.options.getPeerId(), this.options.getTerm(), request.getCommittedIndex(), request.getPrevLogIndex(), request.getPrevLogTerm(), nextSendingIndex, request.getEntriesCount()});
        }

        this.statInfo.runningState = Replicator.RunningState.APPENDING_ENTRIES;
        this.statInfo.firstLogIndex = rb.getPrevLogIndex() + 1L;
        this.statInfo.lastLogIndex = rb.getPrevLogIndex() + (long)rb.getEntriesCount();
        final Recyclable recyclable = dataBuf;
        final int v = this.version;
        final long monotonicSendTimeMs = Utils.monotonicMs();
        final int seq = this.getAndIncrementReqSeq();
        ++this.appendEntriesCounter;
        Future<Message> rpcFuture = null;

        try {
            rpcFuture = this.rpcService.appendEntries(this.options.getPeerId().getEndpoint(), request, -1, new RpcResponseClosureAdapter<RpcRequests.AppendEntriesResponse>() {
                public void run(Status status) {
                    RecycleUtil.recycle(recyclable);
                    Replicator.onRpcReturned(Replicator.this.id, Replicator.RequestType.AppendEntries, status, request, this.getResponse(), seq, v, monotonicSendTimeMs);
                }
            });
        } catch (Throwable var17) {
            RecycleUtil.recycle(dataBuf);
            ThrowUtil.throwException(var17);
        }

        this.addInflight(Replicator.RequestType.AppendEntries, nextSendingIndex, request.getEntriesCount(), request.getData().size(), seq, rpcFuture);
        return true;
    }
}

这个方法会构造一个RpcRequests.AppendEntriesRequest请求，并通过rpcService.appendEntries方法发送出去，并实现了异步回调，用于了解其他nacos节点接收日志的情况，同时后面还会调用上面提到的com.alibaba.nacos.naming.core.v2.service.impl.PersistentClientOperationServiceImpl#onApply

本站总访问量：次本站总访问数：人
本站由 Peng Tong 使用 Stellar 1.29.1 主题创建。
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议，转载请注明出处。