Moderen dağıtık sistemler dünyasında gecikme düşmandır. İşbirlikli bir belge düzenleyici, canlı bir ticaret paneli veya bir IoT telemetri sistemi oluşturuyor olun, standart istek-cevap kalıpları genellikle yetersiz kalır. İşte tam da burada gRPC'nin iki yönlü akışı, sürekli ve düşük gecikmeli veri akışı sağlayarak öne çıkar. Ancak, bu akışları üretim ortamına hazır hale getirmek için kimlik doğrulama, günlük kaydı ve hata işleme için güçlü kesiciler entegre etmeniz gerekir. Bu yazıda, iki yönlü akışı nasıl uygulayacağınızı ve Go'da kesicilerden yararlanarak sağlam bir gerçek zamanlı veri senkronizasyon motoru oluşturmayı keşfedeceğiz.
İki Yönlü Akışın Anlaşılması
Sunucu taraflı veya istemci taraflı akışın aksine, iki yönlü akış hem istemcinin hem de sunucunun mesajları bağımsız ve eşzamanlı olarak göndermesine ve almasına olanak tanır. Bu, verinin istemci ile merkezi sunucu arasında senkronize edilmesi gibi her iki yönde de veri akışının olduğu senaryolar için idealdir. Go'da bu, mesajları eşzamanlı olarak okuma ve yazma için kanallar sağlayan grpc.ServerStream aracılığıyla yönetilir.
Bunu göstermek için, yapılandırma güncellemelerini senkronize eden basit bir protobuf servisi tanımlayalım. SyncConfig RPC'si, istemciden gelen bir dizi ConfigUpdate mesajını kabul edecek ve SyncStatus mesajlarından oluşan bir akış döndürecektir.
Go'da İki Yönlü Akışın Uygulanması
İki yönlü akışın temel zorluğu, bağlantının yaşam döngüsünü yönetmek ve eşzamanlı okuma/yazma işlemlerini ele almaktır. Go'da, her iki yönün de düzgün şekilde işlendiğinden emin olmak için sync.WaitGroup kullanabiliriz. Aşağıda sunucu tarafı işleyicisinin pratik bir uygulaması yer almaktadır.
func (s *server) SyncConfig(stream pb.ConfigService_SyncConfigServer) error {
var wg sync.WaitGroup
wg.Add(2)
// 1. İstemciden gelen mesajları işle
go func() {
defer wg.Done()
for {
update, err := stream.Recv()
if err == io.EOF {
return
}
if err != nil {
log.Printf("Güncelleme alınırken hata: %v", err)
return
}
// Yapılandırma güncellemesini işle
s.processUpdate(update)
// İstemciye onay gönder
status := &pb.SyncStatus{
Status: pb.SyncStatus_ACCEPTED,
Timestamp: time.Now().Unix(),
}
if err := stream.Send(status); err != nil {
log.Printf("Durum gönderilirken hata: %v", err)
return
}
}
}()
// 2. Giden mesajları işle (örn. sunucu başlatmalı yayımlar)
go func() {
defer wg.Done()
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
broadcast := &pb.SyncStatus{
Status: pb.SyncStatus_HEARTBEAT,
Message: "Sunucu senkronizasyonu aktif",
}
if err := stream.Send(broadcast); err != nil {
return
}
case <-stream.Context().Done():
return
}
}
}()
// Her iki goroutine'in de bitmesini bekle
wg.Wait()
return nil
}
Bu uygulama, kritik bir kalıbı vurgular: okuma döngüsünü yazma döngüsünden ayırmak. Bir sync.WaitGroup kullanarak, gelen güncellemelerin işlenmesi ve kalp atışlarının gönderilmesi tamamlanana kadar sunucunun akışı kapatmadığından emin oluyoruz.
Kesicilerin Gücü
Akış mantığı veri akışını ele alırken, kesiciler çapraz kesen endişeleri (cross-cutting concerns) ele alır. gRPC'de kesiciler, RPC işleyicisinin etrafını saran ve ana iş mantığı yürütülmeden önce ve sonra isteği veya yanıtı incelemenizi veya değiştirmenize olanak tanıyan işlevlerdir. Bu, her milisaniyenin önemli olduğu ve her bağlantının güvence altına alınması gereken gerçek zamanlı sistemler için hayati önem taşır.
İki yönlü akışların süresini izleyen basit bir günlük kaydı kesicisi uygulayalım. Bu, uzun ömürlü bağlantıların sağlığını izlemeye yardımcı olur.
Günlük Kaydı İçin Kesici Uygulama
Go'da kesiciler genellikle tekli (unary) veya akış kesicileridir. Bir akışlı RPC ile uğraştığımız için bir StreamServerInterceptor ihtiyacımız var.
func loggingInterceptor(ctx context.Context, req interface{}, info *grpc.StreamServerInfo, handler grpc.StreamHandler) (interface{}, error) {
start := time.Now()
log.Printf("%s akışı başlatıldı", info.FullMethod)
// Zincirdeki bir sonraki işleyiciyi çağır
stream, err := handler(ctx, req)
if err != nil {
log.Printf("%s akışı başarısız oldu: %v", info.FullMethod, err)
return nil, err
}
// Akışın kapanırken günlüğünü kaydetmek için sar
wrappedStream := &loggingStreamWrapper{
Stream: stream.(grpc.ServerStream),
method: info.FullMethod,
start: start,
}
return nil, wrappedStream.SendAndClose(nil)
}
type loggingStreamWrapper struct {
grpc.ServerStream
method string
start time.Time
}
func (w *loggingStreamWrapper) SendMsg(m interface{}) error {
err := w.ServerStream.SendMsg(m)
if err != nil {
log.Printf("%s içinde mesaj gönderilirken hata: %v", w.method, err)
}
return err
}
func (w *loggingStreamWrapper) RecvMsg(m interface{}) error {
err := w.ServerStream.RecvMsg(m)
if err != nil && err != io.EOF {
log.Printf("%s içinde mesaj alınırken hata: %v", w.method, err)
}
return err
}
Yukarıdaki örnek basitleştirilmiştir. Üretim ortamında, metrikleri doğru bir şekilde yakalamak için akışı daha kapsamlı bir şekilde sarmalamanız gerekir. Ancak, bu mantığın akış yaşam döngüsüne nasıl enjekte edileceğini göstermektedir.
Akışların ve Kesicilerin Entegrasyonu
Hepsini bir araya getirmek için, kesiciyi gRPC sunucu örneğini oluştururken kaydettirirsiniz. Kesici, SyncConfig yöntemini otomatik olarak saracak ve iş mantığınızı karmaşıklaştırmadan tüm iki yönlü bağlantıların performansı ve durumu hakkında görünürlük sağlayacaktır.
opts := []grpc.ServerOption{
grpc.UnaryInterceptor(loggingInterceptor), // Not: Akış için grpc.StreamInterceptor kullanın
grpc.StreamInterceptor(loggingInterceptor),
}
s := grpc.NewServer(opts...)
pb.RegisterConfigServiceServer(s, &server{})
Sonuç
Go gRPC'de iki yönlü akış uygulamak, gerçekten gerçek zamanlı uygulamalar için potansiyeli ortaya çıkarır. Okuma ve yazma döngülerini ayırarak ve sync.WaitGroup gibi standart Go ilkeleroyle eşzamanlılığı yöneterek, güvenilir ve yüksek verimli veri boru hatları oluşturabilirsiniz. Ayrıca, kesicileri entegre etmek, bu karmaşık akışların gözlemlenebilir, güvenli ve sürdürülebilir olmasını sağlar. Daha sofistike gerçek zamanlı sistemler oluşturmaya geçtikçe, bu kalıpları ustalaşmak, sorunsuz bir kullanıcı deneyimi sunmak için hayati önem taşıyacaktır.