kratos 使用 kafka

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/segmentio/kafka-go"
)

// Kafka 配置
var (
  brokerAddress := "192.168.3.125:9092" // Kafka 服务器地址
  topic := "my-topic"                   // Kafka 主题
)

func main() {
	// 创建一个 Kafka Writer，用于发送消息
	writer := kafka.NewWriter(kafka.WriterConfig{
		Brokers:  []string{brokerAddress}, // Kafka 服务器地址
		Topic:    topic,                   // Kafka 主题
		Balancer: &kafka.LeastBytes{},     // 负载均衡策略（按最小字节数分配）
	})

	defer writer.Close() // 程序结束时关闭 Writer

	// 发送消息
	for i := 0; i < 10; i++ {
		msg := fmt.Sprintf("Message %d", i) // 消息内容
		err := writer.WriteMessages(context.Background(), kafka.Message{
			Key:   []byte(fmt.Sprintf("Key-%d", i)), // 消息的 Key（可选）
			Value: []byte(msg),                      // 消息的 Value
		})
		if err != nil {
			log.Fatalf("Failed to write message: %v", err)
		}
		fmt.Printf("Produced message: %s\n", msg)
		time.Sleep(1 * time.Second) // 每秒发送一条消息
	}
}

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/segmentio/kafka-go"
)

func main() {
	// Kafka 配置
	topic := "my-topic"                   // Kafka 主题
	brokerAddress := "192.168.3.125:9092" // Kafka 服务器地址
	groupID := "my-group"                 // 消费者组 ID

	// 创建一个 Kafka Reader，用于消费消息
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  []string{brokerAddress}, // Kafka 服务器地址
		Topic:    topic,                   // Kafka 主题
		GroupID:  groupID,                 // 消费者组 ID
		MinBytes: 10e3,                    // 最小读取字节数（10KB）
		MaxBytes: 10e6,                    // 最大读取字节数（10MB）
	})

	defer reader.Close() // 程序结束时关闭 Reader

	// 消费消息
	for {
		msg, err := reader.ReadMessage(context.Background())
		if err != nil {
			log.Fatalf("Failed to read message: %v", err)
		}
		fmt.Printf("Consumed message: topic=%s partition=%d offset=%d key=%s value=%s\n",
			msg.Topic, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
		time.Sleep(1 * time.Second) // 模拟处理消息的延迟
	}
}

syntax = "proto3";

package helloworld.v1;

import "google/api/annotations.proto";

option go_package = "kratos-kafka2/api/helloworld/v1;v1";
option java_multiple_files = true;
option java_package = "dev.kratos.api.helloworld.v1";
option java_outer_classname = "HelloworldProtoV1";

// The greeting service definition.
service Greeter {
  // Sends a greeting
  rpc CreateUser (Request) returns (Reply) {
    option (google.api.http) = {
      post: "/v1/user/register",
      body: "*"
    };
  }
}

// The request message containing the user's name.
message Request {
  string name = 1;
  string email = 2;
  string password = 3;
  uint32 id = 4;
}

// The response message containing the greetings
message Reply {
  uint32 id = 1;
}

message Data {

  message Kafka {
    repeated string brokers = 1;
    string topic = 2;
    string group_id = 3;
  }
  Kafka kafka = 3;
}

data:
  kafka:
    brokers:
      - 192.168.3.125:9092
    topic: my-topic
    group_id: create-account

package biz

import (
	"context"
	"encoding/json"
	"fmt"

	v1 "kratos-kafka2/api/helloworld/v1"

	"github.com/go-kratos/kratos/v2/errors"
	"github.com/go-kratos/kratos/v2/log"
)

var (
	// ErrUserNotFound is user not found.
	ErrUserNotFound = errors.NotFound(v1.ErrorReason_USER_NOT_FOUND.String(), "user not found")
)

// Greeter is a Greeter model.
type Greeter struct {
	Hello string
}

type User struct {
	ID       int64
	Email    string
	Name     string
	Password string
}

// GreeterRepo is a Greater repo.
type GreeterRepo interface {
	Create(ctx context.Context, user *User) (int64, error)
	KafkaSendMessage(ctx context.Context, key []byte, value []byte) (err error)

	Save(context.Context, *Greeter) (*Greeter, error)
	Update(context.Context, *Greeter) (*Greeter, error)
	FindByID(context.Context, int64) (*Greeter, error)
	ListByHello(context.Context, string) ([]*Greeter, error)
	ListAll(context.Context) ([]*Greeter, error)
}

// GreeterUsecase is a Greeter usecase.
type GreeterUsecase struct {
	repo GreeterRepo
	log  *log.Helper
}

// NewGreeterUsecase new a Greeter usecase.
func NewGreeterUsecase(repo GreeterRepo, logger log.Logger) *GreeterUsecase {
	return &GreeterUsecase{repo: repo, log: log.NewHelper(logger)}
}

func (uc *GreeterUsecase) CreateUser(ctx context.Context, user *User) (id int64, err error) {
	fmt.Printf("CreateUser:%+v\n", user)
	id, err = uc.repo.Create(ctx, user)
	if err != nil {
		return
	}
	if id > 0 {
		fmt.Printf("CreateUser:%+v\n", user)
		var b []byte
		b, err = json.Marshal(user)
		if err != nil {
			return
		}
		err = uc.repo.KafkaSendMessage(ctx, []byte(user.Name), b)
		if err != nil {
			return
		}
	}
	return
}

// CreateGreeter creates a Greeter, and returns the new Greeter.
func (uc *GreeterUsecase) CreateGreeter(ctx context.Context, g *Greeter) (*Greeter, error) {
	uc.log.WithContext(ctx).Infof("CreateGreeter: %v", g.Hello)

	return uc.repo.Save(ctx, g)
}

package data

import (
	"context"
	"github.com/segmentio/kafka-go"
	"kratos-kafka2/internal/conf"

	"github.com/go-kratos/kratos/v2/log"
	"github.com/google/wire"
)

// ProviderSet is data providers.
var ProviderSet = wire.NewSet(NewData, NewGreeterRepo, NewKafkaProducer, NewKafkaConsumer)

// Data .
type Data struct {
	kp *KafkaProducer
	kc *KafkaConsumer
}

// NewData .
func NewData(c *conf.Data, logger log.Logger, kp *KafkaProducer, kc *KafkaConsumer) (*Data, func(), error) {
	cleanup := func() {
		log.NewHelper(logger).Info("closing the data resources")
	}
	return &Data{kp, kc}, cleanup, nil
}

type KafkaProducer struct {
	writer *kafka.Writer
}

func NewKafkaProducer(c *conf.Data) *KafkaProducer {
	brokers := c.Kafka.Brokers
	topic := c.Kafka.Topic
	writer := &kafka.Writer{
		Addr:     kafka.TCP(brokers...),
		Topic:    topic,
		Balancer: &kafka.LeastBytes{},
	}
	return &KafkaProducer{writer: writer}
}

func (p *KafkaProducer) SendMessage(ctx context.Context, key, value []byte) error {
	err := p.writer.WriteMessages(ctx, kafka.Message{
		Key:   key,
		Value: value,
	})
	if err != nil {
		return err
	}
	log.Infof("sent message: %s", string(value))
	return nil
}

func (p *KafkaProducer) Close() error {
	return p.writer.Close()
}

type KafkaConsumer struct {
	reader *kafka.Reader
}

func NewKafkaConsumer(c *conf.Data) *KafkaConsumer {
	brokers := c.Kafka.Brokers
	topic := c.Kafka.Topic
	groupId := c.Kafka.GroupId
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers: brokers,
		Topic:   topic,
		GroupID: groupId,
	})
	return &KafkaConsumer{
		reader: reader,
	}
}

func (c *KafkaConsumer) Start(ctx context.Context) {
	for {
		msg, err := c.reader.ReadMessage(ctx)
		if err != nil {
			return
		}
		log.Debugf("key=%s || value=%s", string(msg.Key), string(msg.Value))
	}
}

func (c *KafkaConsumer) Close() error {
	return c.reader.Close()
}

package data

import (
	"context"
	"github.com/go-kratos/kratos/v2/log"
	"time"
)

func (u *greeterRepo) KafkaSendMessage(ctx context.Context, key []byte, value []byte) (err error) {
	defer u.data.kp.Close()
	// 设置超时时间
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	err = u.data.kp.SendMessage(ctx, key, value)
	if err != nil {
		log.Errorf("KafkaSendMessage() || err=%v", err)
		return
	}
	return
}

package service

import (
	"context"

	v1 "kratos-kafka2/api/helloworld/v1"
	"kratos-kafka2/internal/biz"
)

// GreeterService is a greeter service.
type GreeterService struct {
	v1.UnimplementedGreeterServer

	uc *biz.GreeterUsecase
}

// NewGreeterService new a greeter service.
func NewGreeterService(uc *biz.GreeterUsecase) *GreeterService {
	return &GreeterService{uc: uc}
}

func (s *GreeterService) CreateUser(ctx context.Context, req *v1.Request) (*v1.Reply, error) {
	id, err := s.uc.CreateUser(ctx, &biz.User{
		ID:       int64(req.Id),
		Email:    req.Email,
		Name:     req.Name,
		Password: req.Password,
	})
	if err != nil {
		return nil, err
	}
	return &v1.Reply{Id: uint32(id)}, err
}