go-zero と Kafka+Avro の実践: 高性能対話型データ処理システムの構築-Golang-php.cn

go-zero と Kafka+Avro の実践: 高性能対話型データ処理システムの構築

王林

リリース： 2023-06-23 09:04:35

オリジナル

861 人が閲覧しました

近年、ビッグデータと活発なオープンソースコミュニティの台頭により、ますます多くの企業が増大するデータニーズを満たすために高性能の対話型データ処理システムを探し始めています。このテクノロジーアップグレードの波の中で、go-zero と Kafka Avro はますます多くの企業に注目され、採用されています。

go-zero は、Golang 言語に基づいて開発されたマイクロサービスフレームワークです。高いパフォーマンス、使いやすさ、拡張の容易さ、メンテナンスの容易さの特徴を持ち、企業が効率的なマイクロサービスアプリケーションを迅速に構築できるように設計されています。システム。その急速な成長は、Golang 自体の優れたパフォーマンスと高い開発効率に加え、go-zero チームの継続的な反復と最適化によるものです。

Kafka は、Apache によって開発された分散ストリーム処理システムであり、高可用性と高スループットの特性を備えており、現在のビッグデータエコシステムで最も人気のあるメッセージキューの 1 つです。 Avro は、Apache によって開発されたデータシリアル化ツールです。データストリームをバイナリ形式に変換することで、データ圧縮と送信効率を向上させることができます。また、データ形式のアップグレードと変換もサポートできます。

この記事では、go-zero と Kafka Avro を組み合わせて、高性能な対話型データ処理システムを構築する方法を紹介します。具体的な実際のプロセスは次のとおりです。

Kafka クライアントの統合

まず、Kafka クライアントを go-zero サービスに統合する必要があります。 go-zero は、Kafka と簡単に対話できる Kafka パッケージを提供します。

Kafka パッケージをプロジェクトに導入し、構成ファイルで Kafka パラメーターを構成して、Kafka との接続とデータ対話を実現するだけです。以下は Kafka の構成例です。

[kafka]
addrs = ["localhost:9092"]
version = "2.0.0"
maxMessageBytes = 10000000

ログイン後にコピー

特定のビジネスロジックでは、Kafka が提供するプロデューサー API とコンシューマー API を使用してデータを送受信できます。以下は、Kafka プロデューサーの例です。

var (
    topic = "test"
)

func (s *Service) Produce(msg []byte) error {
    p, err := kafka.NewProducer(s.cfg.Kafka)
    if err != nil {
        return err
    }
    defer p.Close()

    return p.Send(context.TODO(), &kafka.Message{
        Key:   []byte(topic),
        Value: msg,
    })
}

ログイン後にコピー

上の例では、「test」という名前の Kafka トピックを作成し、Produce メソッドが呼び出されると、データがトピックに送信されます。

統合された Avro シリアル化

次に、シリアル化および逆シリアル化のためにデータを Avro 形式に変換する必要があります。 go-zero は Avro パッケージを提供し、コード生成をサポートします。スキーマファイルを定義することで、Avro データをエンコードおよびデコードするための対応する Go コードを生成できます。

以下は Avro Schema の構成例です:

{
    "namespace": "com.example",
    "type": "record",
    "name": "User",
    "fields": [
        {
            "name": "name",
            "type": "string"
        },
        {
            "name": "age",
            "type": "int"
        }
    ]
}

ログイン後にコピー

次のコマンドを実行すると、対応する Go ファイルが自動生成されます:

$ go run github.com/gogo/protobuf/protoc-gen-gogofaster --proto_path=./ example.proto --gogofaster_out

ログイン後にコピー

生成された Go ファイルには、 Avro フィールド型と対応する Go データ型の間のマッピング関係を確認することで、データのシリアル化と逆シリアル化を実現できます。

対話型データ処理システムの構築

Kafka と Avro を統合した後、高性能の対話型データ処理システムの構築を開始できます。 Kafka をデータストレージセンターとして使用し、その中に複数のパーティションを確立して、データの分散ストレージと処理を実現できます。

パーティションごとにコンシューマグループを作成して、データの並列処理と負荷分散を実現できます。同時に、go-zero が提供するコルーチンプールと同期チャネルを使用して、データ処理の同時実行パフォーマンスを最適化できます。

以下は対話型データ処理システムの例です:

// 创建消费组
group, err := kafka.NewGroup(s.cfg.Kafka, "test", kafka.WithGroupID("test-group"))
if err != nil {
    return nil, err
}
// 创建消费者
consumer, err := group.NewConsumer(context.Background(), []string{"test"})
if err != nil {
    return nil, err
}
// 启动并发协程
for i := 0; i < s.cfg.WorkerNum; i++ {
    go func() {
        for {
            select {
                // 从同步通道中获取新消息
                case msg := <-msgs:
                    if err := s.processMsg(msg); err != nil {
                        log.Errorf("failed to process message(%v): %v", msg.Value, err)
                    }
                }
        }
    }()
}
// 消费数据
for {
    m, err := consumer.FetchMessage(context.Background())
    if err != nil {
        log.Errorf("failed to fetch message: %v", err)
        continue
    }
    // 将新消息发送到同步通道中
    msgs <- m
}

ログイン後にコピー

上の例では、コンシューマグループ「test-group」を作成し、対応するコンシューマを作成しました。処理中、まず複数の同時コルーチンを開始して、データの並列処理を実現します。新しいメッセージを受信すると、それを同期チャネルに送信し、非同期処理にコルーチンプールを利用します。

上記の構築により、go-zero、Kafka、Avro を統合し、高性能対話型データ処理システムを実装することに成功しました。このようなシステムを利用することで、大量のデータを簡単に扱えるようになり、データの処理や分析の効率が向上します。

以上がgo-zero と Kafka+Avro の実践: 高性能対話型データ処理システムの構築の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。