RabbitMQ RPC

1、回调队列
2、Correlation Id
3、总结
4、整合

在RabbitMQ 工作队列中，我们学习了如何通过工作队列在多个工作者之间分配耗时任务。

但如果我们需要在远端电脑上执行函数并等待结果呢？嗯，这将是一个完全不同的故事。这种模式被称为远程过程调用或RPC。

本章节，我们将使用RabbitMQ来构建一个RPC系统：一个客户端和一个可扩展的RPC服务端。因为我们没有真实的耗时任务来进行分发，所以我们创建一个斐波那契数列函数来模拟RPC服务。

RPC注意事项 尽管RPC在计算中是非常常见的一种模式，但它也饱受争议。当程序员不知道调用的时本地函数还是慢的RPC服务时，问题就会出现。。像这样的混乱会导致系统不可预测，并给调试增加不必要的复杂性。滥用RPC不仅不会简化软件，还会导致像意大利面一样难以维护的代码。牢记一点，在使用RPC前先考虑以下建议：

确保哪个函数是本地调用，哪个是RPC调用

写好文档，明确各组件之间的关系

处理错误场景。当RPC长时间掉线，客户端应该如何处理？如果存疑，请避免使用RPC。如果可以，你应该使用异步管道 – 而不是RPC的阻塞，结果被异步推送到下一个计算阶段。

1、回调队列

通常，使用RabbitMQ进行RPC是很容易的。客户端发送请求消息，服务端返回响应。为了接收响应，我们需要随请求一起发送“回调”队列地址。我们可以使用默认队列：

q, err := ch.QueueDeclare(
	"",    // name
	false, // durable
	false, // delete when unused
	true,  // exclusive
	false, // noWait
	nil,   // arguments
)

err = ch.Publish(
	"",          // exchange
	"rpc_queue", // routing key
	false,       // mandatory
	false,       // immediate
	amqp.Publishing{
		ContentType:   "text/plain",
		CorrelationId: corrId,
		ReplyTo:       q.Name,
		Body:          []byte(strconv.Itoa(n)),
	})

消息属性 AMQP 0-9-1 协议预定义了一组包含14个与消息一起使用的属性。大多数属性很少使用，除了以下几个：

persistent：标记消息持久化（true）或临时的（false）。在第二节有介绍过。

content_type：用于描述编码的 MIME 类型。例如，对于经常使用的 JSON 编码，最好将此属性设置为：application/json。

reply_to：通常用来命名回调队列。

correlation_id：用于将RPC响应与请求关联。

2、Correlation Id

在上面介绍的方法中，我们建议为每个 RPC 请求创建一个回调队列。这是非常低效的，但幸运的是有一个更好的方法 – 让我们为每个客户端创建一个回调队列。

这就引发了一个新的问题：接收到响应的队列不知该响应对应的是哪个请求。这就是correlation_id属性出场的时候了。我们为每一个队列设置一个唯一的值。之后，从回调队列中接收消息时，我们检查这个属性，基于这个值我们就能将响应与请求匹配在一起。如果接收到了一个未知的correlation_id，那我们可以安全地遗弃它 – 因为它不是我们的请求。

你可能会问，为什么我们要忽略回调队列中的未知消息，而不是失败报错呢？这是因为服务端可能会存在竞争条件。尽管不太可能，但RPC服务可能会在发送完结果，但还未发送消息确认之前就挂了。如果发生了这种情况，RPC服务重启后会再次处理该请求。这就是我们为什么要优雅地处理重复响应。通常情况下RPC是等幂的。

3、总结

我们的RPC工作流程如下：

客户端启动时，会创建一个匿名的独占队列。
对于RPC请求，客户端发送消息时带有两个属性：reply_to，设置回调队列；correlation_id，为每个请求设置唯一的值。
请求发送给rpc_queue队列。
RPC服务等待接收队列中的消息。当有消息出现，它处理消息并将带有结果的消息通过reply_to的队列返回给客户端。
客户端等待回调队列中的消息。当有消息出现，它检查correlation_id属性。如果跟请求中的值匹配，就将响应返回给程序。

4、整合

Fibonacci函数：

func fib(n int) int {
	if n == 0 {
		return 0
	} else if n == 1 {
		return 1
	} else {
		return fib(n-1) + fib(n-2)
	}
}

我们的Fibonacci函数只接收一个整数输入（不要期望它处理很大的数字，这只是一个非常低效的递归实现）。

RPC服务端rpc_server.go代码如下：

package main

import (
	"log"
	"strconv"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func fib(n int) int {
	if n == 0 {
		return 0
	} else if n == 1 {
		return 1
	} else {
		return fib(n-1) + fib(n-2)
	}
}

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare(
		"rpc_queue", // name
		false,       // durable
		false,       // delete when unused
		false,       // exclusive
		false,       // no-wait
		nil,         // arguments
	)
	failOnError(err, "Failed to declare a queue")

	err = ch.Qos(
		1,     // prefetch count
		0,     // prefetch size
		false, // global
	)
	failOnError(err, "Failed to set QoS")

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		false,  // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")

	forever := make(chan bool)

	go func() {
		for d := range msgs {
			n, err := strconv.Atoi(string(d.Body))
			failOnError(err, "Failed to convert body to integer")

			log.Printf(" [.] fib(%d)", n)
			response := fib(n)

			err = ch.Publish(
				"",        // exchange
				d.ReplyTo, // routing key
				false,     // mandatory
				false,     // immediate
				amqp.Publishing{
					ContentType:   "text/plain",
					CorrelationId: d.CorrelationId,
					Body:          []byte(strconv.Itoa(response)),
				})
			failOnError(err, "Failed to publish a message")

			d.Ack(false)
		}
	}()

	log.Printf(" [*] Awaiting RPC requests")
	<-forever
}

服务端的代码很简单：

跟之前一样，先建立连接、通过，然后声明队列。
我们可能会执行多个服务进程。为了在多个服务器上平均分配负载，我们需要在通道上设置prefetch属性。
使用Channel.Consume从队列中获取消息。然后启动协程，在协程中处理工作，并返回结果。

RPC客户端rpc_client.go代码如下：

package main

import (
	"log"
	"math/rand"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func randomString(l int) string {
	bytes := make([]byte, l)
	for i := 0; i < l; i++ {
		bytes[i] = byte(randInt(65, 90))
	}
	return string(bytes)
}

func randInt(min int, max int) int {
	return min + rand.Intn(max-min)
}

func fibonacciRPC(n int) (res int, err error) {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare(
		"",    // name
		false, // durable
		false, // delete when unused
		true,  // exclusive
		false, // noWait
		nil,   // arguments
	)
	failOnError(err, "Failed to declare a queue")

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")

	corrId := randomString(32)

	err = ch.Publish(
		"",          // exchange
		"rpc_queue", // routing key
		false,       // mandatory
		false,       // immediate
		amqp.Publishing{
			ContentType:   "text/plain",
			CorrelationId: corrId,
			ReplyTo:       q.Name,
			Body:          []byte(strconv.Itoa(n)),
		})
	failOnError(err, "Failed to publish a message")

	for d := range msgs {
		if corrId == d.CorrelationId {
			res, err = strconv.Atoi(string(d.Body))
			failOnError(err, "Failed to convert body to integer")
			break
		}
	}

	return
}

func main() {
	rand.Seed(time.Now().UTC().UnixNano())

	n := bodyFrom(os.Args)

	log.Printf(" [x] Requesting fib(%d)", n)
	res, err := fibonacciRPC(n)
	failOnError(err, "Failed to handle RPC request")

	log.Printf(" [.] Got %d", res)
}

func bodyFrom(args []string) int {
	var s string
	if (len(args) < 2) || os.Args[1] == "" {
		s = "30"
	} else {
		s = strings.Join(args[1:], " ")
	}
	n, err := strconv.Atoi(s)
	failOnError(err, "Failed to convert arg to integer")
	return n
}

启动服务：

go run rpc_server.go
# => [x] Awaiting RPC requests

客户端发送fibonacci计算请求：

go run rpc_client.go 30
# => [x] Requesting fib(30)

上面的RPC服务实现并不是唯一的，但它有以下几个优点：

如果RPC服务太慢了，那你可以再启动一个来进行扩展。
客户端侧，RPC只需要发送和接收一条消息。因此，对于单个RPC请求，RPC客户端只需要一次网络往返。

我们代码还是非常简单的，并没有尝试解决更复杂（也更重要）的问题，比如：

如果服务端没有运行，客户端该如何处理？
客户端要设置RPC超时吗？
如果服务端出现故障并引发异常，要发送给客户端吗？
处理前进行入参检查（例如检查边界、类型）。

在实验过程中，management UI对于队列查看是很有帮助的。

声明：本作品采用署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0)进行许可，使用时请注明出处。
Author: mengbin92
Github: mengbin92
cnblogs: 恋水无意