{"id":46911,"date":"2022-07-22T22:49:42","date_gmt":"2023-10-20T01:36:29","guid":{"rendered":"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/"},"modified":"2024-01-15T12:13:56","modified_gmt":"2024-01-15T04:13:56","slug":"%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af","status":"publish","type":"post","link":"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/","title":{"rendered":"\u53d1\u9001Kafka\u6d88\u606f"},"content":{"rendered":"

\u597d\u5427\uff0c\u90a3\u4e48\u6211\u60f3\u8981\u4ece\u5e94\u7528\u7a0b\u5e8f\u4e2d\u4f7f\u7528Kafka\u3002
\n\u5177\u4f53\u6765\u8bf4\uff0c\u6211\u60f3\u8981\u5c06\u4ee5POST\u65b9\u6cd5\u4fdd\u5b58TODO\u7684\u8fc7\u7a0b\u5f02\u6b65\u5316\u3002<\/p>\n

\u4eca\u5929\u6211\u4eec\u6765\u770b\u4e00\u4e0b\u8be5\u6d3b\u52a8\u7684\u53d1\u9001\u90e8\u5206\u3002<\/p>\n

UW\u5b9e\u9a8c\u5ba4\/\u57fa\u5e95<\/h2>\n

\u6211\u4eec\u516c\u53f8\u8fd8\u53d1\u5e03\u4e86\u4e00\u4e2a\u7528\u4e8e\u62bd\u8c61\u5316\u6d88\u606f\u7684Go\u5e93\u3002\u5373\u4f7f\u60a8\u4f7f\u7528\u7684\u662f\u9664Kafka\u4e4b\u5916\u7684\u6d41\u5904\u7406\u5668\uff0c\u4e5f\u53ef\u4ee5\u7528\u76f8\u540c\u7684\u65b9\u5f0f\u8fdb\u884c\u5b9e\u73b0\u3002\u672c\u6b21\u8ba9\u6211\u4eec\u4e00\u8d77\u6765\u770b\u770b\u8fd9\u4e2a\u5e93\u7684\u7528\u6cd5\u3002<\/p>\n

substrate.\u65b0\u7684\u540c\u6b65\u6d88\u606f\u63a5\u6536\u70b9()<\/h2>\n

\u5728\u8fd9\u4e2asubstrate\u5e93\u4e2d\uff0c\u9ed8\u8ba4\u60c5\u51b5\u4e0b\u5f02\u6b65\u5316\u5b8c\u6210\u6d88\u606f\u5904\u7406\u7684\u4fe1\u53f7\uff0c\u4ee5\u63d0\u9ad8\u6027\u80fd\u3002\u4f46\u662f\uff0c\u8fd9\u6b21\u6211\u60f3\u8981\u4f7f\u7528\u540c\u6b65\u7248\u672c\u7684substrate.NewSynchronousMessageSink()\u7684API\u3002
\n\u5728main.go\u4e2d\uff0c\u6211\u4f1a\u5b9a\u4e49\u4e00\u4e2a\u540d\u4e3a`initializeKafkaSink`\u7684\u65b9\u6cd5\u6765\u8bbe\u7f6e\u6d88\u606fsink\uff0c\u5982\u4e0b\u6240\u793a\u3002<\/p>\n

func<\/span> initialiseKafkaSink<\/span>(<\/span>version<\/span>,<\/span> brokers<\/span>,<\/span> topic<\/span> *<\/span>string<\/span>,<\/span> keyFunc<\/span> func<\/span>(<\/span>substrate<\/span>.<\/span>Message<\/span>)<\/span> []<\/span>byte<\/span>)<\/span> (<\/span>substrate<\/span>.<\/span>SynchronousMessageSink<\/span>,<\/span> error<\/span>)<\/span> {<\/span>\r\n    sink<\/span>,<\/span> err<\/span> :=<\/span> kafka<\/span>.<\/span>NewAsyncMessageSink<\/span>(<\/span>kafka<\/span>.<\/span>AsyncMessageSinkConfig<\/span>{<\/span>\r\n        Brokers<\/span>:<\/span> strings<\/span>.<\/span>Split<\/span>(<\/span>*<\/span>brokers<\/span>,<\/span> \",\"<\/span>),<\/span>\r\n        Topic<\/span>:<\/span>   *<\/span>topic<\/span>,<\/span>\r\n        KeyFunc<\/span>:<\/span> keyFunc<\/span>,<\/span>\r\n        Version<\/span>:<\/span> *<\/span>version<\/span>,<\/span>\r\n    })<\/span>\r\n    if<\/span> err<\/span> !=<\/span> nil<\/span> {<\/span>\r\n        return<\/span> nil<\/span>,<\/span> err<\/span>\r\n    }<\/span>\r\n\r\n    return<\/span> substrate<\/span>.<\/span>NewSynchronousMessageSink<\/span>(<\/span>sink<\/span>),<\/span> nil<\/span>\r\n}<\/span>\r\n<\/code><\/pre>\n
\u8c03\u7528\u65b9\u7684\u611f\u89c9\u5982\u4e0b\u6240\u793a\u3002<\/p>\n
unc<\/span> main<\/span>()<\/span> {<\/span>\r\n    ...<\/span>\r\n\r\n    app<\/span>.<\/span>Action<\/span> =<\/span> func<\/span>()<\/span> {<\/span>\r\n        ...<\/span>\r\n\r\n        gSrv<\/span> :=<\/span> initialiseGRPCServer<\/span>(<\/span>newServer<\/span>(<\/span>store<\/span>))<\/span>\r\n\r\n        actionSink<\/span>,<\/span> err<\/span> :=<\/span> initialiseKafkaSink<\/span>(<\/span>sinkKafkaVersion<\/span>,<\/span> sinkBrokers<\/span>,<\/span> actionTopic<\/span>,<\/span> actionKeyFunc<\/span>)<\/span>\r\n        if<\/span> err<\/span> !=<\/span> nil<\/span> {<\/span>\r\n            log<\/span>.<\/span>Fatalln<\/span>(<\/span>\"init action event kafka sink:\"<\/span>,<\/span> err<\/span>)<\/span>\r\n        }<\/span>\r\n        defer<\/span> actionSink<\/span>.<\/span>Close<\/span>()<\/span>\r\n\r\n        errCh<\/span> :=<\/span> make<\/span>(<\/span>chan<\/span> error<\/span>,<\/span> 2<\/span>)<\/span>\r\n\r\n        ...<\/span>\r\n    }<\/span>\r\n\r\n    ...<\/span>\r\n}<\/span>\r\n<\/code><\/pre>\n
\u6211\u5011\u5c0d\u65bc Kafka \u7684\u7248\u672c\u4ee5\u53ca\u5176\u4ed6\u5fc5\u8981\u7684\u74b0\u5883\u8b8a\u6578\u5b9a\u7fa9\u5982\u4e0b\u3002<\/p>\n
sinkKafkaVersion<\/span> :=<\/span> app<\/span>.<\/span>String<\/span>(<\/span>cli<\/span>.<\/span>StringOpt<\/span>{<\/span>\r\n    Name<\/span>:<\/span>   \"sink-kafka-version\"<\/span>,<\/span>\r\n    Desc<\/span>:<\/span>   \"sink kafka version\"<\/span>,<\/span>\r\n    EnvVar<\/span>:<\/span> \"SINK_KAFKA_VERSION\"<\/span>,<\/span>\r\n})<\/span>\r\nsinkBrokers<\/span> :=<\/span> app<\/span>.<\/span>String<\/span>(<\/span>cli<\/span>.<\/span>StringOpt<\/span>{<\/span>\r\n    Name<\/span>:<\/span>   \"sink-brokers\"<\/span>,<\/span>\r\n    Desc<\/span>:<\/span>   \"kafka sink brokers\"<\/span>,<\/span>\r\n    EnvVar<\/span>:<\/span> \"SINK_BROKERS\"<\/span>,<\/span>\r\n    Value<\/span>:<\/span>  \"localhost:9092\"<\/span>,<\/span>\r\n})<\/span>\r\n\r\nactionTopic<\/span> :=<\/span> app<\/span>.<\/span>String<\/span>(<\/span>cli<\/span>.<\/span>StringOpt<\/span>{<\/span>\r\n    Name<\/span>:<\/span>   \"action-topic\"<\/span>,<\/span>\r\n    Desc<\/span>:<\/span>   \"action topic\"<\/span>,<\/span>\r\n    EnvVar<\/span>:<\/span> \"ACTION_TOPIC\"<\/span>,<\/span>\r\n    Value<\/span>:<\/span>  \"qiita.action\"<\/span>,<\/span>\r\n})<\/span>\r\n<\/code><\/pre>\n
\u540e\u7eed\u6211\u4eec\u4f1a\u8be6\u7ec6\u8ba8\u8bba\u7684\u662f\u4f20\u7ed9initializeKafkaSink\u7684actionKeyFunc\u51fd\u6570\u3002<\/p>\n
\u4e8b\u4ef6\u7684\u5b9a\u4e49<\/h2>\n
\u6211\u5011\u73fe\u5728\u60f3\u8981\u5b9a\u7fa9\u4e00\u500b\u540d\u70baCreateTodoActionEvent\u7684\u4e8b\u4ef6\uff0c\u9019\u500b\u4e8b\u4ef6\u5c07\u6703\u88ab\u5be6\u969b\u767c\u9001\u51fa\u53bb\u3002<\/p>\n
syntax<\/span> =<\/span> \"proto3\"<\/span>;<\/span>\r\n\r\npackage<\/span> event<\/span>;<\/span>\r\n\r\nmessage<\/span> CreateTodoActionEvent<\/span> {<\/span>\r\n    string<\/span> id<\/span> =<\/span> 1<\/span>;<\/span>\r\n    string<\/span> title<\/span> =<\/span> 2<\/span>;<\/span>\r\n    string<\/span> description<\/span> =<\/span> 3<\/span>;<\/span>\r\n}<\/span>\r\n<\/code><\/pre>\n
\u5728proto\/service.proto\u4e2d\u5b9a\u4e49\u4e86Todo\u6d88\u606f\uff0c\u56e0\u6b64\u53ef\u4ee5\u91cd\u7528\u5b83\uff0c\u4f46\u7531\u4e8e\u7f16\u8bd1\u6709\u4e9b\u9ebb\u70e6\uff0c\u6240\u4ee5\u6211\u5c06\u5176\u4f5c\u4e3a\u91cd\u6784\u5143\u7d20\u7559\u4e0b\u3002<\/p>\n
\u539f\u578b\u4efb\u52a1\u7684\u5185\u5bb9\u5982\u4e0b\u3002<\/p>\n
.PHONY<\/span>:<\/span> protos<\/span>\r\nprotos<\/span>:<\/span>\r\n    mkdir<\/span> -pv<\/span> $(GENERATED_DIR)<\/span> $(GENERATED_SERVICE_DIR)<\/span> $(GENERATED_EVENT_DIR)<\/span> $(GENERATED_ENVELOPE_DIR)<\/span>\r\n    protoc<\/span> \\<\/span>\r\n        -I<\/span> $(PROTO_DIR)<\/span> \\<\/span>\r\n        -I $(GOPATH)\/src<\/span>:<\/span>$(GOPATH)\/src\/github.com\/gogo\/protobuf\/protobuf <\/span>\\\r\n        --gogoslick_out=plugins=grpc:$(GENERATED_SERVICE_DIR) <\/span>\\\r\n        service.proto<\/span>\r\n    protoc<\/span> \\<\/span>\r\n        -I<\/span> $(PROTO_DIR)<\/span> \\<\/span>\r\n        -I $(GOPATH)\/src<\/span>:<\/span>$(GOPATH)\/src\/github.com\/gogo\/protobuf\/protobuf <\/span>\\\r\n        --gogoslick_out=paths=source_relative:$(GENERATED_EVENT_DIR) <\/span>\\\r\n        event.proto<\/span>\r\n    protoc<\/span> \\<\/span>\r\n        -I<\/span> $(PROTO_DIR)<\/span> \\<\/span>\r\n        -I $(GOPATH)\/src<\/span>:<\/span>$(GOPATH)\/src\/github.com\/gogo\/protobuf\/protobuf <\/span>\\\r\n        --gogoslick_out=paths=source_relative<\/span>,<\/span>$(ENVELOPE_PROTO_MAPPINGS):$(GENERATED_ENVELOPE_DIR) <\/span>\\\r\n        envelope.proto<\/span>\r\n<\/code><\/pre>\n
\u5173\u952e\u51fd\u6570\u7684\u5b9e\u73b0<\/h2>\n
\u8fd9\u662f\u4e00\u4e2a\u540d\u4e3aactionKeyFunc\u7684\u51fd\u6570\uff0c\u4e4b\u524d\u6211\u4eec\u63a8\u8fdf\u4e86\u5b83\u3002
\n\u8fd9\u4e2a\u51fd\u6570\u7528\u4e8e\u5b9a\u4e49\u4e8b\u4ef6\u7684\u952e\u3002
\n\u5728Kafka\u4e2d\uff0c\u5171\u4eab\u76f8\u540c\u952e\u7684\u6d88\u606f\u5c06\u8fdb\u5165\u540c\u4e00\u4e2a\u5206\u533a\uff0c\u8fd9\u662f\u4e00\u4e2a\u975e\u5e38\u91cd\u8981\u7684\u6027\u8d28\uff0c\u6240\u4ee5\u6211\u4eec\u4e0d\u80fd\u5ffd\u89c6\u4f7f\u7528\u4ec0\u4e48\u4f5c\u4e3a\u952e\u7684\u95ee\u9898\u3002
\n\u5728\u4e4b\u524d\u7684\u4e8b\u4ef6\u5b9a\u4e49\u4e2d\uff0c\u6211\u4eec\u6dfb\u52a0\u4e86\u4e00\u4e2aid\u5b57\u6bb5\uff0c\u6240\u4ee5\u8fd9\u6b21\u6211\u4eec\u60f3\u8981\u4f7f\u7528\u5b83\u4f5c\u4e3a\u952e\u3002<\/p>\n
func<\/span> actionKeyFunc<\/span>(<\/span>msg<\/span> substrate<\/span>.<\/span>Message<\/span>)<\/span> []<\/span>byte<\/span> {<\/span>\r\n    var<\/span> env<\/span> envelope<\/span>.<\/span>Event<\/span>\r\n    if<\/span> err<\/span> :=<\/span> proto<\/span>.<\/span>Unmarshal<\/span>(<\/span>msg<\/span>.<\/span>Data<\/span>(),<\/span> &<\/span>env<\/span>);<\/span> err<\/span> !=<\/span> nil<\/span> {<\/span>\r\n        panic<\/span>(<\/span>err<\/span>)<\/span>\r\n    }<\/span>\r\n\r\n    if<\/span> types<\/span>.<\/span>Is<\/span>(<\/span>env<\/span>.<\/span>Payload<\/span>,<\/span> &<\/span>event<\/span>.<\/span>CreateTodoActionEvent<\/span>{})<\/span> {<\/span>\r\n        var<\/span> ev<\/span> event<\/span>.<\/span>CreateTodoActionEvent<\/span>\r\n        if<\/span> err<\/span> :=<\/span> types<\/span>.<\/span>UnmarshalAny<\/span>(<\/span>env<\/span>.<\/span>Payload<\/span>,<\/span> &<\/span>ev<\/span>);<\/span> err<\/span> !=<\/span> nil<\/span> {<\/span>\r\n            panic<\/span>(<\/span>err<\/span>)<\/span>\r\n        }<\/span>\r\n\r\n        return<\/span> []<\/span>byte<\/span>(<\/span>ev<\/span>.<\/span>Id<\/span>)<\/span>\r\n    }<\/span>\r\n\r\n    panic<\/span>(<\/span>\"unknown event\"<\/span>)<\/span>\r\n}<\/span>\r\n<\/code><\/pre>\n
\u5728\u8fd9\u91cc\uff0c\u53ea\u9700\u8981\u4e00\u4e2a\u9009\u9879\uff0c\u5bf9\u4ee5\u4e0b\u5185\u5bb9\u8fdb\u884c\u4e2d\u6587\u7684\u672c\u571f\u5316\u6539\u5199\uff1a
\n– github.com\/gogo\/protobuf\/proto
\n– github.com\/gogo\/protobuf\/types<\/p>\n
\u8fd9\u91cc\u5305\u62ec\uff1a
\n– github.com\/gogo\/protobuf\/proto
\n– github.com\/gogo\/protobuf\/types<\/p>\n
\u4f60\u53ef\u4ee5\u4f7f\u7528\u8fd9\u4e24\u4e2a\u5305\u6765\u4ece\u4fe1\u5c01\u4e2d\u53d6\u51fa\u4e8b\u4ef6\uff0c\u4e5f\u53ef\u4ee5\u786e\u8ba4\u8fd9\u4e2a\u8fc7\u7a0b\u3002<\/p>\n
substrate.Message\u7684\u5b9e\u73b0\u65b9\u5f0f<\/h2>\n
\u6211\u60f3\u5f00\u59cb\u5b9e\u73b0\u6d88\u606f\u53d1\u9001\u90e8\u5206\u7684\u529f\u80fd\uff0c\u4f46\u5728\u6b64\u4e4b\u524d\u9700\u8981\u4e00\u4e2a\u5fc5\u8981\u7684\u6b65\u9aa4\u3002
\n\u6d88\u606f\u53d1\u9001\u7684APIPublishMessage()\u51fd\u6570\u63a5\u53d7substrate.Message\u63a5\u53e3\u4f5c\u4e3a\u53c2\u6570\u3002
\n\u56e0\u6b64\uff0c\u6211\u4eec\u9700\u8981\u51c6\u5907\u4e00\u4e2a\u5b9e\u73b0\u8be5\u63a5\u53e3\u7684\u7c7b\u578b\u3002<\/p>\n
\u53ea\u9700\u4e00\u4e2a\u5fc5\u8981\u7684\u65b9\u6cd5\u3002<\/p>\n
type<\/span> Message<\/span> interface<\/span> {<\/span>\r\n    Data<\/span>()<\/span> []<\/span>byte<\/span>\r\n}<\/span>\r\n<\/code><\/pre>\n
\u5b9a\u4e49\u5982\u4e0b\u7684\u7c7b\u578b\u3002<\/p>\n
type<\/span> message<\/span> []<\/span>byte<\/span>\r\n\r\nfunc<\/span> (<\/span>m<\/span> message<\/span>)<\/span> Data<\/span>()<\/span> []<\/span>byte<\/span> {<\/span>\r\n    return<\/span> m<\/span>\r\n}<\/span>\r\n<\/code><\/pre>\n
\u53d1\u9001\u6d88\u606f<\/h2>\n
\u597d\u7684\uff0c\u90a3\u4e48\u73b0\u5728\u662f\u6d88\u606f\u53d1\u9001\u5904\u7406\u3002
\nserver.CreateTodo()\u65b9\u6cd5\u5c06\u5982\u4e0b\u6240\u793a\u3002<\/p>\n
func<\/span> (<\/span>s<\/span> *<\/span>server<\/span>)<\/span> CreateTodo<\/span>(<\/span>ctx<\/span> context<\/span>.<\/span>Context<\/span>,<\/span> req<\/span> *<\/span>service<\/span>.<\/span>CreateTodoRequest<\/span>)<\/span> (<\/span>*<\/span>service<\/span>.<\/span>CreateTodoResponse<\/span>,<\/span> error<\/span>)<\/span> {<\/span>\r\n    todoID<\/span> :=<\/span> uuid<\/span>.<\/span>New<\/span>()<\/span>.<\/span>String<\/span>()<\/span>\r\n    ev<\/span> :=<\/span> &<\/span>event<\/span>.<\/span>CreateTodoActionEvent<\/span>{<\/span>\r\n        Id<\/span>:<\/span>          todoID<\/span>,<\/span>\r\n        Title<\/span>:<\/span>       req<\/span>.<\/span>Todo<\/span>.<\/span>Title<\/span>,<\/span>\r\n        Description<\/span>:<\/span> req<\/span>.<\/span>Todo<\/span>.<\/span>Description<\/span>,<\/span>\r\n    }<\/span>\r\n\r\n    any<\/span>,<\/span> err<\/span> :=<\/span> types<\/span>.<\/span>MarshalAny<\/span>(<\/span>ev<\/span>)<\/span>\r\n    if<\/span> err<\/span> !=<\/span> nil<\/span> {<\/span>\r\n        return<\/span> nil<\/span>,<\/span> err<\/span>\r\n    }<\/span>\r\n\r\n    env<\/span> :=<\/span> envelope<\/span>.<\/span>Event<\/span>{<\/span>\r\n        Id<\/span>:<\/span>        uuid<\/span>.<\/span>New<\/span>()<\/span>.<\/span>String<\/span>(),<\/span>\r\n        Timestamp<\/span>:<\/span> types<\/span>.<\/span>TimestampNow<\/span>(),<\/span>\r\n        Payload<\/span>:<\/span>   any<\/span>,<\/span>\r\n    }<\/span>\r\n\r\n    b<\/span>,<\/span> err<\/span> :=<\/span> proto<\/span>.<\/span>Marshal<\/span>(<\/span>&<\/span>env<\/span>)<\/span>\r\n    if<\/span> err<\/span> !=<\/span> nil<\/span> {<\/span>\r\n        return<\/span> nil<\/span>,<\/span> err<\/span>\r\n    }<\/span>\r\n\r\n    if<\/span> err<\/span> :=<\/span> s<\/span>.<\/span>sink<\/span>.<\/span>PublishMessage<\/span>(<\/span>ctx<\/span>,<\/span> message<\/span>(<\/span>b<\/span>));<\/span> err<\/span> !=<\/span> nil<\/span> {<\/span>\r\n        return<\/span> nil<\/span>,<\/span> err<\/span>\r\n    }<\/span>\r\n\r\n    return<\/span> &<\/span>service<\/span>.<\/span>CreateTodoResponse<\/span>{<\/span>\r\n        Success<\/span>:<\/span> true<\/span>,<\/span>\r\n        Id<\/span>:<\/span>      todoID<\/span>,<\/span>\r\n    },<\/span> nil<\/span>\r\n}<\/span>\r\n<\/code><\/pre>\n
\u8bf7\u5173\u6ce8\u5c06\u4e8b\u4ef6\u4f5c\u4e3a\u6709\u6548\u8d1f\u8f7d\u5c01\u88c5\u5728\u4fe1\u5c01\u4e2d\u7684\u90e8\u5206\uff0c\u7136\u540e\u5c06\u6574\u4e2a\u5185\u5bb9\u8f6c\u6362\u4e3a[]byte\u8fdb\u884c\u7f16\u7ec4\u7684\u5904\u7406\u3002<\/p>\n
\u670d\u52a1\u5668\u7ed3\u6784\u4f53\u4f9d\u8d56\u4e8esubstrate.SynchrounousMessageSink\u3002<\/p>\n
type<\/span> (<\/span>\r\n    ...<\/span>\r\n\r\n    server<\/span> struct<\/span> {<\/span>\r\n        sink<\/span> substrate<\/span>.<\/span>SynchronousMessageSink<\/span>\r\n    }<\/span>\r\n)<\/span>\r\n<\/code><\/pre>\n
\u6211\u5011\u61c9\u8a72\u5beb\u4e00\u6bb5\u6e2c\u8a66\u50b3\u9001\u8a0a\u606f\u7a0b\u5f0f\u78bc\uff0c\u4e0d\u904e\u9019\u6b21\u6211\u5011\u7a0d\u5fae\u5077\u61f6\u4e86\u3002\u5e0c\u671b\u5728\u672a\u4f86\u7684\u67d0\u4e00\u5929\u80fd\u5920\u63d0\u5230\u5b83\u3002<\/p>\n
\u4f9d\u8d56\u5173\u7cfb\u7684\u6ce8\u5165<\/h2>\n
\u6211\u5011\u9700\u8981\u5728 main.go \u4e2d\u6ce8\u5165\u4ee5\u4e0b\u7684\u4f9d\u8cf4\u95dc\u4fc2\u3002<\/p>\n
diff --git a\/main.go b\/main.go\r\nindex be109e8..907cd41 100644\r\n<\/span>--- a\/main.go\r\n<\/span>+++ b\/main.go\r\n<\/span>@@ -104,8 +104,6 @@<\/span> func main() {\r\n        }\r\n        defer lis.Close()\r\n\r\n-       gSrv := initialiseGRPCServer(newServer(nil))\r\n-\r\n<\/span>        actionSink, err := initialiseKafkaSink(sinkKafkaVersion, sinkBrokers, actionTopic, actionKeyFunc)\r\n        if err != nil {\r\n            log.Fatalln(\"init payment account kafka sink:\", err)\r\n@@ -122,6 +120,9 @@<\/span> func main() {\r\n        }()\r\n\r\n        var wg sync.WaitGroup\r\n+\r\n+       gSrv := initialiseGRPCServer(newServer(actionSink))\r\n+\r\n<\/span>        wg.Add(1)\r\n        go func() {\r\n            defer wg.Done()\r\n<\/code><\/pre>\n
\u6211\u8ba4\u4e3a\u5728\u53d1\u9001\u6d88\u606f\u4e4b\u524d\uff0c\u5b9e\u9645\u4e0a\u9700\u8981\u901a\u8fc7\u9a8c\u8bc1\u7b49\u624b\u6bb5\u6765\u68c0\u67e5\u662f\u5426\u53ef\u4ee5\u771f\u6b63\u4fdd\u5b58\uff0c\u4f46\u603b\u7684\u6765\u8bf4\uff0c\u8fd9\u5df2\u7ecf\u5b8c\u6210\u4e86\u6d88\u606f\u7684\u53d1\u9001\u90e8\u5206\u3002<\/p>\n
\n
\u53d1\u73b0\u6bd4\u60f3\u8c61\u4e2d\u7684\u4efb\u52a1\u66f4\u591a\uff0c\u6240\u4ee5\u4e8b\u60c5\u8fdb\u884c\u5730\u6709\u4e9b\u4ed3\u4fc3\u3002\u975e\u5e38\u62b1\u6b49\u3002<\/p>\n
\u7531\u4e8e\u6d88\u606f\u53d1\u9001\u8fc7\u7a0b\u5df2\u7ecf\u5b8c\u6210\uff0c\u63a5\u4e0b\u6765\u6211\u4eec\u5e0c\u671b\u7ee7\u7eed\u5904\u7406\u63a5\u6536\u6d88\u606f\u5e76\u5c06\u5176\u4fdd\u5b58\u5230\u6570\u636e\u5e93\u4e2d\u3002<\/p>\n","protected":false},"excerpt":{"rendered":"
\u597d\u5427\uff0c\u90a3\u4e48\u6211\u60f3\u8981\u4ece\u5e94\u7528\u7a0b\u5e8f\u4e2d\u4f7f\u7528Kafka\u3002 \u5177\u4f53\u6765\u8bf4\uff0c\u6211\u60f3\u8981\u5c06\u4ee5POST\u65b9\u6cd5\u4fdd\u5b58TODO\u7684\u8fc7\u7a0b\u5f02\u6b65\u5316\u3002 \u4eca\u5929 […]<\/p>\n","protected":false},"author":7,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-46911","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"yoast_head":"\n\u53d1\u9001Kafka\u6d88\u606f - Blog - Silicon Cloud<\/title>\n<meta name=\"robots\" content=\"index, follow, max-snippet:-1, max-image-preview:large, max-video-preview:-1\" \/>\n<link rel=\"canonical\" href=\"https:\/\/www.silicloud.com\/zh\/blog\/\u53d1\u9001kafka\u6d88\u606f\/\" \/>\n<meta property=\"og:locale\" content=\"zh_CN\" \/>\n<meta property=\"og:type\" content=\"article\" \/>\n<meta property=\"og:title\" content=\"\u53d1\u9001Kafka\u6d88\u606f\" \/>\n<meta property=\"og:description\" content=\"\u597d\u5427\uff0c\u90a3\u4e48\u6211\u60f3\u8981\u4ece\u5e94\u7528\u7a0b\u5e8f\u4e2d\u4f7f\u7528Kafka\u3002 \u5177\u4f53\u6765\u8bf4\uff0c\u6211\u60f3\u8981\u5c06\u4ee5POST\u65b9\u6cd5\u4fdd\u5b58TODO\u7684\u8fc7\u7a0b\u5f02\u6b65\u5316\u3002 \u4eca\u5929 […]\" \/>\n<meta property=\"og:url\" content=\"https:\/\/www.silicloud.com\/zh\/blog\/\u53d1\u9001kafka\u6d88\u606f\/\" \/>\n<meta property=\"og:site_name\" content=\"Blog - Silicon Cloud\" \/>\n<meta property=\"article:published_time\" content=\"2023-10-20T01:36:29+00:00\" \/>\n<meta property=\"article:modified_time\" content=\"2024-01-15T04:13:56+00:00\" \/>\n<meta name=\"author\" content=\"\u79d1, \u9896\" \/>\n<meta name=\"twitter:card\" content=\"summary_large_image\" \/>\n<meta name=\"twitter:label1\" content=\"\u4f5c\u8005\" \/>\n\t<meta name=\"twitter:data1\" content=\"\u79d1, \u9896\" \/>\n\t<meta name=\"twitter:label2\" content=\"\u9884\u8ba1\u9605\u8bfb\u65f6\u95f4\" \/>\n\t<meta name=\"twitter:data2\" content=\"3 \u5206\" \/>\n<script type=\"application\/ld+json\" class=\"yoast-schema-graph\">{\"@context\":\"https:\/\/schema.org\",\"@graph\":[{\"@type\":\"WebPage\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/\",\"url\":\"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/\",\"name\":\"\u53d1\u9001Kafka\u6d88\u606f - Blog - Silicon Cloud\",\"isPartOf\":{\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#website\"},\"datePublished\":\"2023-10-20T01:36:29+00:00\",\"dateModified\":\"2024-01-15T04:13:56+00:00\",\"author\":{\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/8ca01ba7f7362ad4edb7da206a12f29e\"},\"breadcrumb\":{\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/#breadcrumb\"},\"inLanguage\":\"zh-Hans\",\"potentialAction\":[{\"@type\":\"ReadAction\",\"target\":[\"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/\"]}]},{\"@type\":\"BreadcrumbList\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/#breadcrumb\",\"itemListElement\":[{\"@type\":\"ListItem\",\"position\":1,\"name\":\"\u9996\u9875\",\"item\":\"https:\/\/www.silicloud.com\/zh\/blog\/\"},{\"@type\":\"ListItem\",\"position\":2,\"name\":\"\u53d1\u9001Kafka\u6d88\u606f\"}]},{\"@type\":\"WebSite\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#website\",\"url\":\"https:\/\/www.silicloud.com\/zh\/blog\/\",\"name\":\"Blog - Silicon Cloud\",\"description\":\"\",\"inLanguage\":\"zh-Hans\"},{\"@type\":\"Person\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/8ca01ba7f7362ad4edb7da206a12f29e\",\"name\":\"\u79d1, \u9896\",\"image\":{\"@type\":\"ImageObject\",\"inLanguage\":\"zh-Hans\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/image\/\",\"url\":\"https:\/\/secure.gravatar.com\/avatar\/8a6fb3cc7ba2f69d2189ba532aec4633ea7ed75ac0af162ec367cb3abc0fb2af?s=96&d=mm&r=g\",\"contentUrl\":\"https:\/\/secure.gravatar.com\/avatar\/8a6fb3cc7ba2f69d2189ba532aec4633ea7ed75ac0af162ec367cb3abc0fb2af?s=96&d=mm&r=g\",\"caption\":\"\u79d1, \u9896\"},\"url\":\"https:\/\/www.silicloud.com\/zh\/blog\/author\/keying\/\"},{\"@type\":\"ImageObject\",\"inLanguage\":\"zh-Hans\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/#local-main-organization-logo\",\"url\":\"\",\"contentUrl\":\"\",\"caption\":\"Blog - Silicon Cloud\"}]}<\/script>\n<!-- \/ Yoast SEO Premium plugin. -->","yoast_head_json":{"title":"\u53d1\u9001Kafka\u6d88\u606f - Blog - Silicon Cloud","robots":{"index":"index","follow":"follow","max-snippet":"max-snippet:-1","max-image-preview":"max-image-preview:large","max-video-preview":"max-video-preview:-1"},"canonical":"https:\/\/www.silicloud.com\/zh\/blog\/\u53d1\u9001kafka\u6d88\u606f\/","og_locale":"zh_CN","og_type":"article","og_title":"\u53d1\u9001Kafka\u6d88\u606f","og_description":"\u597d\u5427\uff0c\u90a3\u4e48\u6211\u60f3\u8981\u4ece\u5e94\u7528\u7a0b\u5e8f\u4e2d\u4f7f\u7528Kafka\u3002 \u5177\u4f53\u6765\u8bf4\uff0c\u6211\u60f3\u8981\u5c06\u4ee5POST\u65b9\u6cd5\u4fdd\u5b58TODO\u7684\u8fc7\u7a0b\u5f02\u6b65\u5316\u3002 \u4eca\u5929 […]","og_url":"https:\/\/www.silicloud.com\/zh\/blog\/\u53d1\u9001kafka\u6d88\u606f\/","og_site_name":"Blog - Silicon Cloud","article_published_time":"2023-10-20T01:36:29+00:00","article_modified_time":"2024-01-15T04:13:56+00:00","author":"\u79d1, \u9896","twitter_card":"summary_large_image","twitter_misc":{"\u4f5c\u8005":"\u79d1, \u9896","\u9884\u8ba1\u9605\u8bfb\u65f6\u95f4":"3 \u5206"},"schema":{"@context":"https:\/\/schema.org","@graph":[{"@type":"WebPage","@id":"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/","url":"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/","name":"\u53d1\u9001Kafka\u6d88\u606f - Blog - Silicon Cloud","isPartOf":{"@id":"https:\/\/www.silicloud.com\/zh\/blog\/#website"},"datePublished":"2023-10-20T01:36:29+00:00","dateModified":"2024-01-15T04:13:56+00:00","author":{"@id":"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/8ca01ba7f7362ad4edb7da206a12f29e"},"breadcrumb":{"@id":"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/#breadcrumb"},"inLanguage":"zh-Hans","potentialAction":[{"@type":"ReadAction","target":["https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/"]}]},{"@type":"BreadcrumbList","@id":"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/#breadcrumb","itemListElement":[{"@type":"ListItem","position":1,"name":"\u9996\u9875","item":"https:\/\/www.silicloud.com\/zh\/blog\/"},{"@type":"ListItem","position":2,"name":"\u53d1\u9001Kafka\u6d88\u606f"}]},{"@type":"WebSite","@id":"https:\/\/www.silicloud.com\/zh\/blog\/#website","url":"https:\/\/www.silicloud.com\/zh\/blog\/","name":"Blog - Silicon Cloud","description":"","inLanguage":"zh-Hans"},{"@type":"Person","@id":"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/8ca01ba7f7362ad4edb7da206a12f29e","name":"\u79d1, \u9896","image":{"@type":"ImageObject","inLanguage":"zh-Hans","@id":"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/image\/","url":"https:\/\/secure.gravatar.com\/avatar\/8a6fb3cc7ba2f69d2189ba532aec4633ea7ed75ac0af162ec367cb3abc0fb2af?s=96&d=mm&r=g","contentUrl":"https:\/\/secure.gravatar.com\/avatar\/8a6fb3cc7ba2f69d2189ba532aec4633ea7ed75ac0af162ec367cb3abc0fb2af?s=96&d=mm&r=g","caption":"\u79d1, \u9896"},"url":"https:\/\/www.silicloud.com\/zh\/blog\/author\/keying\/"},{"@type":"ImageObject","inLanguage":"zh-Hans","@id":"https:\/\/www.silicloud.com\/zh\/blog\/%e5%8f%91%e9%80%81kafka%e6%b6%88%e6%81%af\/#local-main-organization-logo","url":"","contentUrl":"","caption":"Blog - Silicon Cloud"}]}},"_links":{"self":[{"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/posts\/46911","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/users\/7"}],"replies":[{"embeddable":true,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/comments?post=46911"}],"version-history":[{"count":2,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/posts\/46911\/revisions"}],"predecessor-version":[{"id":59372,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/posts\/46911\/revisions\/59372"}],"wp:attachment":[{"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/media?parent=46911"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/categories?post=46911"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/tags?post=46911"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}