{"id":10020896,"date":"2025-07-10T14:09:42","date_gmt":"2025-07-10T06:09:42","guid":{"rendered":"https:\/\/itmax.cn\/?p=10020896"},"modified":"2025-07-10T14:09:42","modified_gmt":"2025-07-10T06:09:42","slug":"vivo-pulsar-%e4%b8%87%e4%ba%bf%e7%ba%a7%e6%b6%88%e6%81%af%e5%a4%84%e7%90%86%e5%ae%9e%e8%b7%b5%ef%bc%883%ef%bc%89-kop%e6%8c%87%e6%a0%87%e5%bc%82%e5%b8%b8%e4%bf%ae%e5%a4%8d","status":"publish","type":"post","link":"https:\/\/itmax.cn\/?p=10020896","title":{"rendered":"vivo Pulsar \u4e07\u4ebf\u7ea7\u6d88\u606f\u5904\u7406\u5b9e\u8df5\uff083\uff09-KoP\u6307\u6807\u5f02\u5e38\u4fee\u590d"},"content":{"rendered":"
\n

\u4f5c\u8005\uff1avivo \u4e92\u8054\u7f51\u5927\u6570\u636e\u56e2\u961f- Chen Jianbo<\/p>\n

\u672c\u6587\u662f\u300avivo Pulsar\u4e07\u4ebf\u7ea7\u6d88\u606f\u5904\u7406\u5b9e\u8df5\u300b\u7cfb\u5217\u6587\u7ae0\u7b2c3\u7bc7\u3002<\/p>\n

Pulsar\u662fApache\u57fa\u91d1\u4f1a\u7684\u5f00\u6e90\u5206\u5e03\u5f0f\u6d41\u5904\u7406\u5e73\u53f0\u548c\u6d88\u606f\u4e2d\u95f4\u4ef6\uff0c\u5b83\u5b9e\u73b0\u4e86Kafka\u7684\u534f\u8bae\uff0c\u53ef\u4ee5\u8ba9\u4f7f\u7528Kafka API\u7684\u5e94\u7528\u76f4\u63a5\u8fc1\u79fb\u81f3Pulsar\uff0c\u8fd9\u4f7f\u5f97Pulsar\u5728Kafka\u751f\u6001\u7cfb\u7edf\u4e2d\u66f4\u52a0\u5bb9\u6613\u88ab\u63a5\u53d7\u548c\u4f7f\u7528\u3002KoP\u63d0\u4f9b\u4e86\u4eceKafka\u5230Pulsar\u7684\u65e0\u7f1d\u8f6c\u6362\uff0c\u7528\u6237\u53ef\u4ee5\u4f7f\u7528Kafka API\u64cd\u4f5cPulsar\u96c6\u7fa4\uff0c\u4fdd\u7559\u4e86Kafka\u7684\u5e7f\u6cdb\u7528\u6237\u57fa\u7840\u548c\u4e30\u5bcc\u751f\u6001\u7cfb\u7edf\u3002\u5b83\u4f7f\u5f97Pulsar\u53ef\u4ee5\u66f4\u597d\u5730\u4e0eKafka\u8fdb\u884c\u6574\u5408\uff0c\u63d0\u4f9b\u66f4\u597d\u7684\u6d88\u606f\u4f20\u8f93\u6027\u80fd\u3001\u66f4\u5f3a\u7684\u517c\u5bb9\u6027\u53ca\u53ef\u6269\u5c55\u6027\u3002vivo\u5728\u4f7f\u7528Pulsar KoP\u7684\u8fc7\u7a0b\u4e2d\u9047\u5230\u8fc7\u4e00\u4e9b\u95ee\u9898\uff0c\u672c\u7bc7\u4e3b\u8981\u5206\u4eab\u4e00\u4e2a\u5206\u533a\u6d88\u8d39\u6307\u6807\u7f3a\u5931\u7684\u95ee\u9898\u3002<\/p>\n<\/blockquote>\n

\u7cfb\u5217\u6587\u7ae0\uff1a<\/p>\n

    \n
  1. vivo Pulsar\u4e07\u4ebf\u7ea7\u6d88\u606f\u5904\u7406\u5b9e\u8df5\uff081\uff09-\u6570\u636e\u53d1\u9001\u539f\u7406\u89e3\u6790\u548c\u6027\u80fd\u8c03\u4f18<\/a><\/li>\n
  2. vivo Pulsar\u4e07\u4ebf\u7ea7\u6d88\u606f\u5904\u7406\u5b9e\u8df5\uff082\uff09-\u4ece0\u52301\u5efa\u8bbePulsar\u6307\u6807\u76d1\u63a7\u94fe\u8def<\/a><\/li>\n<\/ol>\n

    \u6587\u7ae0\u592a\u957f\uff1f1\u5206\u949f\u770b\u56fe\u6293\u4f4f\u6838\u5fc3\u89c2\u70b9\ud83d\udc47<\/p>\n

    \"\u56fe\u7247\"<\/p>\n

    \u4e00\u3001\u95ee\u9898\u80cc\u666f<\/h2>\n

    \u5728\u4e00\u6b21\u7248\u672c\u7070\u5ea6\u5347\u7ea7\u4e2d\uff0c\u6211\u4eec\u53d1\u73b0\u67d0\u4e2a\u4f7f\u7528KoP\u7684\u4e1a\u52a1topic\u7684\u6d88\u8d39\u901f\u7387\u51fa\u73b0\u4e86\u663e\u8457\u4e0b\u964d\uff0c\u5177\u4f53\u60c5\u51b5\u5982\u4e0b\u56fe\u6240\u793a\uff1a<\/p>\n

    \"\"<\/p>\n

    \u4ec0\u4e48\u539f\u56e0\u5bfc\u81f4\u6b63\u5e38\u7684\u5347\u7ea7\u91cd\u542f\u670d\u52a1\u5668\u4f1a\u51fa\u73b0\u8fd9\u4e2a\u95ee\u9898\u5462\uff1f\u76f4\u63a5\u67e5\u770b\u4e0a\u62a5\u91c7\u96c6\u7684\u6570\u636e\u62a5\u6587\uff1a<\/p>\n

    kop_server_MESSAGE_OUT{group="",partition="0",tenant="kop",topic="persistent:\/\/kop-tenant\/kop-ns\/service-raw-stats"} 3\nkop_server_BYTES_OUT{group="",partition="0",tenant="kop",topic="persistent:\/\/kop-tenant\/kop-ns\/service-raw-stats"} 188\n<\/code><\/pre>\n
    \u6211\u4eec\u770b\u5230\uff0cKoP\u6d88\u8d39\u6307\u6807kop\/_server\/_MESSAGE<\/p>\n
    \/_OUT\u3001kop\/_server\/_BYTES\/_OUT\u662f\u6709\u4e0a\u62a5\u7684\uff0c\u4f46\u6307\u6807\u6570\u636e\u91cc\u7684group\u6807\u7b7e\u53d8\u6210\u4e86\u7a7a\u4e32\uff08\u7f3a\u5c11\u6d88\u8d39\u7ec4\u540d\u79f0\uff09\uff0c\u5206\u533a\u7684\u6d88\u8d39\u6307\u6807\u5c31\u65e0\u6cd5\u5c55\u793a\u4e86\u3002\u662f\u4ec0\u4e48\u539f\u56e0\u5bfc\u81f4\u4e86\u6d88\u8d39\u7ec4\u540d\u79f0\u7f3a\u5931\uff1f<\/p>\n
    \u4e8c\u3001\u95ee\u9898\u5206\u6790<\/h2>\n
    1\u3001\u627e\u5230\u95ee\u9898\u4ee3\u7801<\/strong><\/p>\n
    \u6211\u4eec\u53bb\u627e\u4e0b\u8fd9\u4e2a\u6d88\u8d39\u7ec4\u540d\u79f0\u662f\u5728\u54ea\u91cc\u83b7\u53d6\u7684\uff0c\u662f\u5426\u903b\u8f91\u5b58\u5728\u4ec0\u4e48\u95ee\u9898\u3002\u6839\u636edruid\u4e2d\u7684kop\/_subscription\u5bf9\u5e94\u7684\u6d88\u8d39\u6307\u6807kop\/_server_<\/p>\n
    MESSAGE\/_OUT\u3001kop\/_server\/_BYTES\/_OUT\uff0c\u627e\u5230\u76f8\u5173\u4ee3\u7801\u5982\u4e0b\uff1a<\/p>\n
    private void handleEntries(final List<Entry> entries,\n                               final TopicPartition topicPartition,\n                               final FetchRequest.PartitionData partitionData,\n                               final KafkaTopicConsumerManager tcm,\n                               final ManagedCursor cursor,\n                               final AtomicLong cursorOffset,\n                               final boolean readCommitted) {\n....\n        \/\/ \u5904\u7406\u6d88\u8d39\u6570\u636e\u65f6\uff0c\u83b7\u53d6\u6d88\u8d39\u7ec4\u540d\u79f0\n        CompletableFuture<String> groupNameFuture = requestHandler\n                .getCurrentConnectedGroup()\n                .computeIfAbsent(clientHost, clientHost -> {\n                    CompletableFuture<String> future = new CompletableFuture<>();\n                    String groupIdPath = GroupIdUtils.groupIdPathFormat(clientHost, header.clientId());\n                    requestHandler.getMetadataStore()\n                            .get(requestHandler.getGroupIdStoredPath() + groupIdPath)\n                            .thenAccept(getResultOpt -> {\n                                if (getResultOpt.isPresent()) {\n                                    GetResult getResult = getResultOpt.get();\n                                    future.complete(new String(getResult.getValue() == null\n                                            ? new byte[0] : getResult.getValue(), StandardCharsets.UTF_8));\n                                } else {\n                                    \/\/ \u4ecezk\u8282\u70b9 \/client_group_id\/xxx \u83b7\u53d6\u4e0d\u5230\u6d88\u8d39\u7ec4\uff0c\u6d88\u8d39\u7ec4\u5c31\u662f\u7a7a\u7684\n                                    future.complete("");\n                                }\n                            }).exceptionally(ex -> {\n                                future.completeExceptionally(ex);\n                                return null;\n                            });\n                    returnfuture;\n                });\n\n        \/\/ this part is heavyweight, and we should not execute in the ManagedLedger Ordered executor thread\n        groupNameFuture.whenCompleteAsync((groupName, ex) -> {\n            if (ex != null) {\n                log.error("Get groupId failed.", ex);\n                groupName = "";\n            }\n.....\n            \/\/ \u83b7\u5f97\u6d88\u8d39\u7ec4\u540d\u79f0\u540e\uff0c\u8bb0\u5f55\u6d88\u8d39\u7ec4\u5bf9\u5e94\u7684\u6d88\u8d39\u6307\u6807\n            decodeResult.updateConsumerStats(topicPartition,\n                    entries.size(),\n                    groupName,\n                    statsLogger);\n<\/code><\/pre>\n
    \u4ee3\u7801\u7684\u903b\u8f91\u662f\uff0c\u4ecerequestHandler\u7684currentConnectedGroup(map)\u4e2d\u901a\u8fc7host\u83b7\u53d6groupName\uff0c\u4e0d\u5b58\u5728\u5219\u901a\u8fc7MetadataStore\uff08\u5e26\u7f13\u5b58\u7684zk\u5b58\u50a8\u5bf9\u8c61\uff09\u83b7\u53d6\uff0c\u5982\u679czk\u7f13\u5b58\u4e5f\u6ca1\u6709\uff0c\u518d\u53d1\u8d77zk\u8bfb\u8bf7\u6c42\uff08\u8def\u5f84\u4e3a\/client\/_group\/_id\/host-clientId\uff09\u3002\u8bfb\u53d6\u5230\u6d88\u8d39\u7ec4\u540d\u79f0\u540e\uff0c\u7528\u5b83\u6765\u66f4\u65b0\u6d88\u8d39\u7ec4\u6307\u6807\u3002\u4ece\u590d\u73b0\u7684\u96c6\u7fa4\u786e\u5b9a\u8d70\u7684\u662f\u8fd9\u4e2a\u5206\u652f\uff0c\u5373\u662f\u4ecemetadataStore(\u5e26\u7f13\u5b58\u7684zk\u5ba2\u6237\u7aef)\u83b7\u53d6\u4e0d\u5230\u5bf9\u5e94zk\u8282\u70b9\/client\/_group\/_id\/xxx\u3002<\/p>\n
    2\u3001\u67e5\u627e\u53ef\u80fd\u5bfc\u81f4zk\u8282\u70b9\/client\/_group\/_id\/xxx\u8282\u70b9\u83b7\u53d6\u4e0d\u5230\u7684\u539f\u56e0<\/strong><\/p>\n
    \u6709\u4e24\u79cd\u53ef\u80fd\u6027\uff1a\u4e00\u662f\u6ca1\u5199\u8fdb\u53bb\uff0c\u4e8c\u662f\u5199\u8fdb\u53bb\u4f46\u662f\u88ab\u5220\u9664\u4e86\u3002<\/p>\n
        @Override\n    protected void handleFindCoordinatorRequest(KafkaHeaderAndRequest findCoordinator,\n                                                CompletableFuture<AbstractResponse> resultFuture) {\n...\n        \/\/ Store group name to metadata store for current client, use to collect consumer metrics.\n        storeGroupId(groupId, groupIdPath)\n                .whenComplete((stat, ex) -> {\n                    if (ex != null) {\n                        \/\/ \/client_group_id\/xxx\u8282\u70b9\u5199\u5165\u5931\u8d25\n                        log.warn("Store groupId failed, the groupId might already stored.", ex);\n                    }\n                    findBroker(TopicName.get(pulsarTopicName))\n                            .whenComplete((node, throwable) -> {\n                                ....\n                            });\n                });\n...\n<\/code><\/pre>\n
    \u4ece\u4ee3\u7801\u770b\u5230\uff0cclientId\u4e0egroupId\u7684\u5173\u8054\u5173\u7cfb\u662f\u901a\u8fc7handleFindCoordinatorRequest\uff08FindCoordinator\uff09\u5199\u8fdb\u53bb\u7684\uff0c\u800c\u4e14\u53ea\u6709\u8fd9\u4e2a\u65b9\u6cd5\u5165\u53e3\u3002\u7531\u4e8e\u6ca1\u6709\u627e\u5230warn\u65e5\u5fd7\uff0c\u6392\u9664\u4e86\u7b2c\u4e00\u79cd\u6ca1\u5199\u8fdb\u53bb\u7684\u53ef\u80fd\u6027\u3002\u770b\u770b\u5220\u9664\u7684\u903b\u8f91\uff1a<\/p>\n
    protected void close(){\n    if (isActive.getAndSet(false)) {\n        ...\n        currentConnectedClientId.forEach(clientId -> {\n            String path = groupIdStoredPath + GroupIdUtils.groupIdPathFormat(clientHost, clientId);\n            \/\/ \u5220\u9664zk\u4e0a\u7684 \/client_group_id\/xxx \u8282\u70b9\n            metadataStore.delete(path, Optional.empty())\n                    .whenComplete((__, ex) -> {\n                        if (ex != null) {\n                            if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {\n                                if (log.isDebugEnabled()) {\n                                    log.debug("The groupId store path doesn't exist. Path: [{}]", path);\n                                }\n                                return;\n                            }\n                            log.error("Delete groupId failed. Path: [{}]", path, ex);\n                            return;\n                        }\n                        if (log.isDebugEnabled()) {\n                            log.debug("Delete groupId success. Path: [{}]", path);\n                        }\n                    });\n        });\n    }\n}\n<\/code><\/pre>\n
    \u5220\u9664\u662f\u5728requsetHandler.close\u65b9\u6cd5\u4e2d\u6267\u884c\uff0c\u4e5f\u5c31\u662f\u8bf4\u8fde\u63a5\u65ad\u5f00\u5c31\u4f1a\u89e6\u53d1zk\u8282\u70b9\u5220\u9664\u3002<\/p>\n
    \u4f46\u6709\u51e0\u4e2a\u7591\u95ee\uff1a<\/strong><\/p>\n