Event flow and event sourcing

package main

import (
"context"
"fmt"
"github.com/segmentio/kafka-go"
)

func main() {
// 设置Kafka生产者以将事件发送到主题
writer := kafka.NewWriter(kafka.WriterConfig{
Brokers: []string{"localhost:9092"},
Topic: "my-topic",
})

// 发送一些事件到主题
writer.WriteMessages(context.Background(),
kafka.Message{
Key: []byte("key1"),
Value: []byte("value1"),
},
kafka.Message{
Key: []byte("key2"),
Value: []byte("value2"),
},
)

// 设置Kafka消费者以从主题读取事件
reader := kafka.NewReader(kafka.ReaderConfig{
Brokers: []string{"localhost:9092"},
Topic: "my-topic",
})

// 从主题读取事件
for {
msg, err := reader.ReadMessage(context.Background())
if err != nil {
break
}
fmt.Printf("Received message: key=%s, value=%sn", string(msg.Key), string(msg.Value))
}
}

package main

import (
"fmt"
)

type Event struct {
Type string
Data interface{}
}

type EventStore struct {
events []Event
}

func (store *EventStore) Append(event Event) {
store.events = append(store.events, event)
}

func (store *EventStore) GetEvents() []Event {
return store.events
}

type Account struct {
idstring
balance int
store *EventStore
}

func NewAccount(id string, store *EventStore) *Account {
return &Account{
id:id,
balance: 0,
store: store,
}
}

func (account *Account) Deposit(amount int) {
event := Event{
Type: "deposit",
Data: amount,
}
account.store.Append(event)
account.balance += amount
}

func (account *Account) Withdraw(amount int) {
if account.balance >= amount {
event := Event{
Type: "withdraw",
Data: amount,
}
account.store.Append(event)
account.balance -= amount
}
}

func (account *Account) GetBalance() int {
return account.balance
}

func main() {
store := &EventStore{}
account := NewAccount("123", store)

account.Deposit(100)
account.Withdraw(50)
account.Deposit(25)

events := store.GetEvents()
for _, event := range events {
switch event.Type {
case "deposit":
amount := event.Data.(int)
fmt.Printf("Deposited %dn", amount)
case "withdraw":
amount := event.Data.(int)
fmt.Printf("Withdrew %dn", amount)
}
}

fmt.Printf("Final balance: %dn", account.GetBalance())
}