[译] 为 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

容器时代志愿者招募