2400字带你把云计算服务架构指令池的搭建和使用一次性搞懂

互推小编 2023-07-04

指令池的搭建与使用

指令池是服务于任务变更的（如取消任务、暂停任务、执行指令等），后端应用程序可以通过指令池把任务变更指令通知到执行该任务的云计算服务软件。

对应的，每个云计算服务软件运行之初都需要在RabbitMQ中创建一个专属的指令消息队列。

云计算服务软件开始执行任务时，需要把专属的指令消息队列名称与任务ID发送给后端应用程序（通过RESTful API），后端应用程序需要记录其对应关系。

后端应用程序发送任务变更指令时，通过默认交换机和指定的指令消息队列名称，把变更指令发送到正在执行该任务的云计算服务软件中。

后端应用程序发送任务变更指令和云计算服务软件领取任务变更指令的流程如图5.26所示。其中，后端应用程序不会同步得知变更指令的执行结果，该执行结果可以体现在任务进度或中间结果当中。

图5.26 后端应用程序发送任务变更指令和云计算服务软件领取任务变更指令的流程

如果需要发送任务变更指令（取消任务等）时，该任务还没被执行（任务停留在任务池中），那么后端应用程序需要先把任务变更指令记录在数据库。

当任务开始被执行后，云计算服务软件会通过RESTful API通知后端应用程序，此时后端应用程序才把任务变更指令发送到指定指令消息队列中。

在一些特殊的运维场景中（如一键暂停或取消所有任务），后端应用程序需要广播共通的变更指令。对应的，可以创建一个广播模式的交换机，且每个后端应用程序都把专属的指令消息队列绑定到该交换机上。这样的话，后端应用程序就可以通过广播模式的交换机一次性地把变更指令发布到所有的指令消息队列中。

除了以上介绍的“任务变更指令发送与领取流程”“任务未开始时需要把任务变更指令滞后发送”“为特殊运维场景增加广播模式的交换机”以外，一般来说，指令消息队列还需要关注以下两个细节：

·指令消息队列的名称需要唯一（随机生成一部分名称），且最好有统一的前缀，例如state.1345132345。

·指令消息队列可设置为断开连接即自动删除，因为每次云计算服务软件启动时都会创建一个新的专属指令消息队列。

指令消息队列搭建与使用的相关代码如代码5.17所示，其中，示例代码是使用C++编写的，而后端应用程序一般是使用Java等开发语言编写的，所以在实际编码中，后端应用程序发送任务消息部分的编码可能会有所区别。

代码5.17　指令消息队列搭建与使用的相关代码

//创建任务消息队列，由云计算服务软件或后端应用程序创建都可以
const char* queueNameStr = "任务消息队列名称";
amqp_boolean_t durable = 1; //设置任务消息队列持久化（重启后消息仍不丢失）
amqp_boolean_t autodelete = 0; //设置任务消息队列不自动删除
amqp_queue_declare(connection, channel, amqp_cstring_bytes
(queueNameStr), 0,
durable, 0, autodelete, amqp_empty_table);//创建广播模式的交换机，由云计算服务软件或后端应用程序创建都可以
const char* exchangeFanout = "用于广播的交换机的名称";
amqp_exchange_declare(connection, channel, amqp_cstring_bytes
(exchangeFanout),
amqp_cstring_bytes("fanout"), 0, 0, 0, 0, amqp_empty_
table);
//后端应用程序相关
const char* exchange = ""; //默认交换机名称（名称为空字符）
const char* routingkey = "指令消息队列名称"; //路由键设置为指令消息队列名称
//设置消息的相关信息
amqp_basic_properties_t props;
props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG |
AMQP_BASIC_DELIVERY_MODE_FLAG; //与下面的配置对应
//消息主体的类型
props.content_type = amqp_cstring_bytes("text/plain");
props.delivery_mode = 2; //持久化消息
//发送消息
const char* messagebody = "发送的指令消息";
amqp_basic_publish(connection, channel, amqp_cstring_bytes(exchange),
amqp_cstring_bytes(routingkey), 0, 0, &props,
amqp_cstring_bytes(messagebody);
//广播共通的指令消息
const char* exchangeFanout = "用于广播的交换机的名称";
const char* messagebody = "发送的任务消息";
amqp_basic_publish(connection, channel, amqp_cstring_bytes
(exchangeFanout),
amqp_cstring_bytes(""), 0, 0, &props,
amqp_cstring_bytes(messagebody);
//云计算服务软件相关
const char* queueNameStr = "指令消息队列名称";
amqp_boolean_t durable = 0; //设置指令消息队列非持久化（重启后消息会丢失）
amqp_boolean_t autodelete = 1; //设置指令消息队列自动删除
amqp_queue_declare(connection, channel, amqp_cstring_bytes
(queueNameStr), 0,
durable, 0, autodelete, amqp_empty_table);
//订阅消息队列，只调用一次即可
amqp_boolean_t noack = 1; //设置为消息自动确认，手动确认也可以
const char* queueNameStr = "指令消息队列名称";
amqp_basic_consume(connection, channel, amqp_cstring_bytes
(queueNameStr),
amqp_empty_bytes, 0, noack, 0,amqp_empty_table);
//获取消息，可以多次调用，无消息时会自动阻塞
amqp_envelope_t envelope; //定义接收消息的变量amqp_maybe_release_buffers(connection); //清理buffers
amqp_consume_message(connection, &envelope, NULL, 0); //获取消息
envelope.message.body.bytes; //消息主体的开始指针（char *）
envelope.message.body.len; //消息主体的长度

指令消息一般为JSON格式，这样能做到灵活配置。后端应用程序在发送任务时，需要把指令消息转换成字符串。而云计算服务在获取到指令消息后，需要把字符串转成JSON格式。指令消息一般包含指令标识、任务ID、指令参数，如代码5.18所示，其中，如果变更指令的执行结果需要通知后端应用程序，则可以在指令消息中加入回调后端应用程序的RESTful API地址等。另外，为防止一些问题，云计算服务软件在执行指令前，需要先判断任务ID，如果任务ID非当前执行的任务对应的ID，则丢弃此指令消息。