@Override
    public StreamObserver<MeterDataCollection> collectBatch(StreamObserver<Commands> responseObserver) {
        return new StreamObserver<MeterDataCollection>() {
            @Override
            public void onNext(MeterDataCollection meterDataCollection) {
                final MeterProcessor processor = processService.createProcessor();
                try (HistogramMetrics.Timer ignored = histogram.createTimer()) {
                    meterDataCollection.getMeterDataList().forEach(processor::read);
                    processor.process();
                } catch (Exception e) {
                    errorCounter.inc();
                    log.error(e.getMessage(), e);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                Status status = Status.fromThrowable(throwable);
                if (Status.CANCELLED.getCode() == status.getCode()) {
                    if (log.isDebugEnabled()) {
                        log.debug(throwable.getMessage(), throwable);
                    }
                    return;
                }
                log.error(throwable.getMessage(), throwable);
            }

            @Override
            public void onCompleted() {
                responseObserver.onNext(Commands.newBuilder().build());
                responseObserver.onCompleted();
            }
        };
    }
Meter protocol receiver, collect and process the meters.
oap-server/server-receiver-plugin/skywalking-meter-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/meter/provider/handler/MeterServiceHandler.java