[譯] 為 Go 語言開發者介紹 NATS

當NATS作為釋出-訂閱引擎時，它提供了三種訊息傳遞模式：

• 釋出-訂閱

• 佇列

• 請求-響應

訊息架構元件

安裝伺服器和客戶端

go get github.com/nats-io/gnatsd

gnatsd

go get github.com/nats-io/go-nats

在Go語言中使用NATS

message ServiceDiscovery {
string order_service_uri = 1;
}

message EventStore {
string aggregate_id = 1;
string aggregate_type = 2;
string event_id = 3;
string event_type = 4;
string event_data = 5;
}

NATS的請求應答示例

func main() {
// Create NATS server connection
natsConnection, _ := nats.Connect(nats.DefaultURL)
log.Println("Connected to " + nats.DefaultURL)

msg, err := natsConnection.Request("Discovery.OrderService", nil, 1000*time.Millisecond)
if err == nil && msg != nil {
orderServiceDiscovery := pb.ServiceDiscovery{}
err := proto.Unmarshal(msg.Data, &orderServiceDiscovery)if err != nil {
log.Fatalf("Error on unmarshal: %v", err)
}
address := orderServiceDiscovery.OrderServiceUri
log.Println("OrderService endpoint found at:", address)
//Set up a connection to the gRPC server.
conn, err := grpc.Dial(address, grpc.WithInsecure())
}
}

import "github.com/nats-io/go-nats"

函式nats.Connect嘗試連線NATS系統，預設的NATS伺服器運在“nats://localhost:4222”。這裡我們使用預設的URL連線伺服器。

natsConnection, _ := nats.Connect(nats.DefaultURL)

NATS通常在專案中傳送一個名為“Discovery.OrderService”的請求以得到應答。

msg, err := natsConnection.Request("Discovery.OrderService", nil, 1000*time.Millisecond)

當對某個專案傳送請求時，你可以傳遞請求資料和超時。這裡我們不提供任何資料，向這個專案傳送請求只是為了接收應答。我們採用Protocol Buffers進行傳送和接收訊息，響應資料被編碼在Go語言結構值中。

orderServiceDiscovery := pb.ServiceDiscovery{}
err := proto.Unmarshal(msg.Data, &orderServiceDiscovery
)if err != nil {
log.Fatalf("Error on unmarshal: %v", err)
}
address := orderServiceDiscovery.OrderServiceUri
log.Println("OrderService endpoint found at:", address)

這是另一個應用的程式碼塊，訂閱專案 “Discovery.OrderService” 提供了對應請求的應答。

清單3:NATS的請求-應答訊息傳遞的響應

var orderServiceUri string
orderServiceUri = viper.GetString("discovery.orderservice")

func main() {
// Create server connection
natsConnection, _ := nats.Connect(nats.DefaultURL)
log.Println("Connected to " + nats.DefaultURL)

natsConnection.Subscribe("Discovery.OrderService", func(m *nats.Msg) {
orderServiceDiscovery := pb.ServiceDiscovery{OrderServiceUri: orderServiceUri}
data, err := proto.Marshal(&orderServiceDiscovery)
if err == nil {
natsConnection.Publish(m.Reply, data)
}
})
// Keep the connection alive
runtime.Goexit()
}

專案 “Discovery.OrderService”參與傳送響應，這裡通過編碼到Protocol Buers以傳送響應資料。

NATS釋出訂閱示例

我極力推薦NATS用於釋出訂閱引擎的pub/sub訊息模型，構建企業級訊息佇列及Go語言搭建的分散式系統。NATS釋出訂閱是一對多通訊系統，專案中一個釋出者傳送一個訊息，專案中所有活躍的訂閱者接收這個訊息。此通訊模型是典型的非同步方式，釋出的訊息被分發到訂閱訊息的處理者。如果沒有處理者，訂閱以非同步模型工作，客戶端有可能被阻塞直到訊息被處理。大多數真實情況下，你可能不需要通訊的非同步方式用於pub/sub通訊。

當建立訂閱者時，你可以同時為它註冊一個佇列名稱。所有具有相同的佇列名稱的訂閱者構成一個佇列組。隨著訊息在被註冊的專案中釋出，佇列組中被隨機選擇的一個訂閱者用於接收訊息。雖然佇列組中有多個訂閱者，但每個訊息只能被一個訂閱者接收，且只能接收一次。當建立訂閱者時，你可以選擇是否註冊佇列名。在佇列組中的訂閱者們，其中一個訂閱者接收訊息，而那些沒有佇列組的訂閱者們，所有訂閱者共同接受這條訊息。有意思的是NATS本身提供佇列甚至它本身也是基於訊息釋出-訂閱模式。

在本文的訊息釋出-訂閱模式，我們建立一個無佇列的訂閱者及一個名“Order.OrdersCreatedQueue”含有多個訂閱者的訂閱組。因此佇列組中的一個訂閱者及其他的訂閱者（無佇列組的）可以接收訊息。釋出者的客戶端是一個gRPC伺服器，當專案“Order.OrderCreated”中建立一個命令時，釋出者將釋出一個訊息，詳見下面的程式碼塊：

清單4:用於NATS釋出訂閱訊息傳遞的釋出者客戶端

const (
aggregate = "Order"
event= "OrderCreated")
// publishOrderCreated publish an event via NATS server

func publishOrderCreated(order *pb.Order) {
// Connect to NATS server
natsConnection, _ := nats.Connect(nats.DefaultURL)
log.Println("Connected to " + nats.DefaultURL)
defer natsConnection.Close()
eventData, _ := json.Marshal(order)
event := pb.EventStore{
AggregateId:order.OrderId,
AggregateType: aggregate,
EventId:uuid.NewV4().String(),
EventType:event,
EventData:string(eventData),
}
subject := "Order.OrderCreated"
data, _ := proto.Marshal(&event)
// Publish message on subject
natsConnection.Publish(subject, data)
log.Println("Published message on subject " + subject)
}

訊息模式

const subject = "Order.>"

func main() {
// Create server connection
natsConnection, _ := nats.Connect(nats.DefaultURL)
log.Println("Connected to " + nats.DefaultURL)
// Subscribe to subject
natsConnection.Subscribe(subject, func(msg *nats.Msg) {
eventStore := pb.EventStore{}
err := proto.Unmarshal(msg.Data, &eventStore)
if err == nil {
// Handle the message
log.Printf("Received message in EventStore service: %+v\n", eventStore)
store := store.EventStore{}
store.CreateEvent(&eventStore)
log.Println("Inserted event into Event Store")
}
})
// Keep the connection alive
runtime.Goexit()
}

使用萬用字元專案“Order.>”訂閱訊息，NATS支援在專案訂閱中使用萬用字元，支援星號字元（*）和大於號（>），也被認為所有的萬用字元被用於通配專案訂閱。萬用字元Order.> will 被匹配為Order.Created，Order.Shipped, Order.Delivered, Order.Delivered.Returned等等。萬用字元Order.* 將被匹配為 Order.Created, Order.Shipped, Order.Delivered等等，而不是 Order.Delivered.Returned。

NATS客戶端的Subscribe功能，當訊息被給定的專案釋出時，訂閱訊息處理非同步接收訊息。 由於訊息由Protocol Buffers編碼所釋出，接收訊息則是通過proto.Unmarshal解碼到Go語言結構體值中。

讓我們新增訂閱者到一個佇列組中，這是訂閱者客戶端從專案中訂閱訊息的程式碼塊：

清單6:訂閱者客戶端佇列組的NATS釋出-訂閱訊息傳遞

const (
queue= "Order.OrdersCreatedQueue"
subject = "Order.OrderCreated"
)
func main() {
// Create server connection
natsConnection, _ := nats.Connect(nats.DefaultURL)
log.Println("Connected to " + nats.DefaultURL)
// Subscribe to subject
natsConnection.QueueSubscribe(subject, queue, func(msg *nats.Msg) {
eventStore := pb.EventStore{}
err := proto.Unmarshal(msg.Data, &eventStore)
if err == nil {
// Handle the message
log.Printf("Subscribed message in Worker 1: %+v\n", eventStore)
}
})
// Keep the connection alive
runtime.Goexit()
}

主題“Order.OrderCreated”的訊息使用名為“Order.OrdersCreatedQueue"的佇列通過QueueSubscribe功能被訂閱。當我們使用同一個佇列名建立多個訂閱者，它們被建立在一個佇列組中，隨機選擇一個訂閱者用來接收訊息。如果只是想通過NATS僅僅用於排隊，可以只通過一個佇列組建立訂閱者。

本文我僅展示了核心NATS平臺的基礎能力，後續我將寫另一篇文章講述NATS流伺服器。

本文中的原始碼可見：https://github.com/shijuvar/gokit/tree/master/examples/grpc-nats

你也可以在twitte r ( @shijucv)  上關注我

原文連結

原文作者：Shiju Varghese

原文連結： https://medium.com/@shijuvar/introducing-nats-to-go-developers-3cfcb98c21d0

容器時代志願者招募