华为云用户手册

场景4：无索引、有索引不走 某一次点查询，Seq Scan扫描需要3767ms，因涉及从4096000条数据中获取8240条数据，符合索引扫描的场景（海量数据中寻找少量数据），在对过滤条件列增加索引后，计划依然是Seq Scan而没有走Index Scan。 对目标表ANALYZE后，计划能够自动选择索引，性能从3s+优化到2ms+，极大降低I/O消耗。 常见场景：行存大表的查询场景，从大量数据中访问极少数据，没走索引扫描而是走顺序扫描，导致I/O效率低，不走索引常见有两种情况： 过滤条件列上没建索引。 有索引但是计划没选索引扫描。 触发因素： 常用过滤条件列没有建索引。 表中数据因执行DML操作后产生数据变化未及时ANALYZE，导致优化器无法选择索引扫描计划，ANALYZE介绍参见ANALYZE。 处理方式： 对行存表常用过滤列增加索引，索引基本设计原则： 索引列选择distinct值多，且常用于过滤条件，过滤条件多时可以考虑建组合索引，组合索引中distinct值多的列排在前面，索引个数不宜超过3个。 大量数据带索引导入会产生大量I/O，如果该表涉及大量数据导入，需严格控制索引个数，建议导入前先将索引删除，导入完成后再重新建索引。 对频繁做DML操作的表，业务中加入及时ANALYZE，主要场景： 表数据从无到有。 表频繁进行INSERT/UPDATE/DELETE。 表数据即插即用，需要立即访问且只访问刚插入的数据。

场景5：无分区、有分区不剪枝 例如某业务表经常使用createtime时间列作为过滤条件获取特定时间数据，对该表设计为分区表后没有走分区剪枝（Selected Partitions数量多），Scan花了701785ms，I/O效率极低。 在增加分区键createtime作为过滤条件后，Partitioned scan走分区剪枝（Selected Partitions数量极少），性能从700s优化到10s，I/O效率极大提升。 常见场景：按照时间存储数据的大表，查询特征大多为访问当天或者某几天的数据，这种情况应该通过分区键进行分区剪枝（只扫描对应少量分区）来极大提升I/O效率，不走分区剪枝常见的情况有： 未设计成分区表。 设计了分区没使用分区键做过滤条件。 分区键做过滤条件时，对列值有函数转换。 触发因素：未合理使用分区表和分区剪枝功能，导致扫描效率低。 处理方式： 对按照时间特征存储和访问的大表设计成分区表。 分区键一般选离散度高、常用于查询过滤条件中的时间类型的字段。 分区间隔一般参考高频的查询所使用的间隔，需要注意的是针对列存表，分区间隔过小（例如按小时）可能会导致小文件过多的问题，一般建议最小间隔为按天。

场景6：行存表求count值 某行存大表频繁全表count（指不带过滤条件或者过滤条件过滤很少数据的count），其中Scan花费43s，持续占用大量I/O，此类作业并发起来后，整体系统I/O持续100%，触发I/O瓶颈，导致整体性能慢。 对比相同数据量的列存表（A-rows均为40960000），列存的Scan只花费14ms，I/O占用极低。 触发因素：行存表因其存储方式的原因，全表scan的效率较低，频繁的对大表全表扫描，导致I/O持续占用。 解决办法： 业务侧审视频繁全表count的必要性，降低全表count的频率和并发度。 如果业务类型符合列存表，则将行存表修改为列存表，提高I/O效率。

场景7：行存表求max值 计算某行存表某列的max值，花费了26772ms，此类作业并发起后，整体系统I/O持续100%，触发I/O瓶颈，导致整体性能慢。 针对max列增加索引后，语句耗时从26s优化到32ms，极大减少I/O消耗。 触发因素：行存表max值逐个scan符合条件的值来计算max，当scan的数据量很大时，会持续消耗I/O。 解决办法：给max列增加索引，凭借btree索引数据有序存储的特征，加速扫描过程，降低I/O消耗。

场景1：列存小CU膨胀 某业务SQL查询出390871条数据需43248ms，分析计划主要耗时在Cstore Scan。 Cstore Scan的详细信息中，每个DN扫描出2w左右的数据，但是扫描了有数据的CU（CUSome）155079个，没有数据的CU（CUNone）156375个，说明当前小CU、未命中数据的CU极多，即CU膨胀严重。 触发因素：对列存表（尤其是分区表）进行高频小批量导入会造成CU膨胀。 处理方法 列存表的数据入库方式修改为攒批入库，单分区单批次入库数据量需大于DN个数*6W。 如果因业务原因无法攒批入库，则需定期VACUUM FULL此类高频小批量导入的列存表。 当小CU膨胀很快时，频繁VACUUM FULL也会消耗大量I/O，甚至加剧整个系统的I/O瓶颈，此场景建议修改列存表为行存表（CU长期膨胀严重的情况下，列存的存储空间优势和顺序扫描性能优势将不复存在）。

场景3：表存储倾斜 例如表Scan的A-time中，max time DN执行耗时6554ms，min time DN耗时0s，DN之间扫描差异超过10倍以上，这种集合Scan的详细信息，基本可以确定为表存储倾斜导致。 通过table_distribution发现所有数据倾斜到了dn_6009单个DN，修改分布列使得表存储分布均匀后，max dn time和min dn time基本维持在相同水平400ms左右，Scan时间从6554ms优化到431ms。 触发因素：分布式场景，表分布列选择不合理会导致存储倾斜，同时导致DN间压力失衡，单DN I/O压力大，整体I/O效率下降。 解决办法：修改表的分布列使表的存储分布均匀，分布列选择原则参见选择分布列。

场景2：脏数据&数据清理 某业务SQL总执行时间2.519s，其中Scan占了2.516s，同时该表的扫描最终只扫描到0条符合条件数据，过滤了20480条数据，即总共扫描了20480+0条数据却消耗了2s+，扫描时间与扫描数据量严重不符，此现象可判断为由于脏数据多从而影响扫描和I/O效率。 查看表脏页率为99%，VACUUM FULL后性能优化到100ms左右。 触发因素：表频繁执行UPDATE/DELETE导致脏数据过多，且长时间未VACUUM FULL清理。 处理方法 对频繁UPDATE/DELETE产生脏数据的表，定期VACUUM FULL，因大表的VACUUM FULL也会消耗大量I/O，因此需要在业务低峰时执行，避免加剧业务高峰期I/O压力。 当脏数据产生很快，频繁VACUUM FULL也会消耗大量I/O，甚至加剧整个系统的I/O瓶颈，这时需要考虑脏数据的产生是否合理。针对频繁DELETE的场景，可以考虑如下方案： 全量DELETE修改为TRUNCATE或者使用临时表替代。 定期DELETE某时间段数据，使用分区表并使用TRUNCATE或DROP分区替代。

原因分析 SQL运行慢可从以下几方面进行分析： 使用EXPLAIN命令查看SQL执行计划，根据执行计划判断是否需要进行SQL调优。 分析查询是否被阻塞，导致语句运行时间过长，可以强制结束有问题的会话。 审视和修改表定义。选择合适的分布列，避免数据倾斜。 分析SQL语句是否使用了不下推的函数，建议更换为支持下推的语法或函数。 对表定期执行VACUUM FULL和ANALYZE操作，可回收已更新或已删除的数据所占据的磁盘空间。 检查表有无索引支撑，建议例行重建索引。 数据库经过多次删除操作后，索引页面上的索引键将被删除，造成索引膨胀。例行重建索引，可有效的提高查询效率。 对业务进行优化，分析能否将大表进行分表设计。

处理方法 8.1.x及以上集群版本，通过pgxc_lock_conflicts视图查看锁冲突情况。 1 SELECT * FROM pgxc_lock_conflicts; 8.0.x及之前集群版本，执行以下SQL查询查看是否有阻塞的SQL语句。 1 2 3 4 5 6 7 8 9 10 11 SELECT w.query as waiting_query, w.pid as w_pid, w.usename as w_user, l.query as locking_query, l.pid as l_pid, l.usename as l_user, n.nspname || '.' || c.relname as tablename from pg_stat_activity w join pg_locks l1 on w.pid = l1.pid and not l1.granted join pg_locks l2 on l1.relation = l2.relation and l2.granted join pg_stat_activity l on l2.pid = l.pid join pg_class c on c.oid = l1.relation join pg_namespace n on n.oid=c.relnamespace where w.waiting; 查询到阻塞的表及模式信息后，请根据实际查询语句ID结束会话： 1 SELECT pgxc_terminate_query(query_id); 或 1 EXECUTE DIRECT ON(cn_name) 'SELECT pg_terminate_backend(pid)'; 这种情况一般是因为业务调度不太合理，建议合理安排各个业务的调度时间。 还可以通过设置GUC参数lockwait_timeout，控制单个锁的最长等待时间，即单个锁的等待超时时间。 lockwait_timeout单位为毫秒（ms），默认值为20分钟。 lockwait_timeout参数属于SUSET类型参数，请参考设置GUC参数中对应的设置方法进行设置。

处理方法 删除该表索引信息。 1 DROP INDEX a_0317_index; 对该表索引进行重建。 1 CREATE INDEX a_0317_index on a_0317(a) local (partition p1_index, partition p2_inde); 查看表定义无报错。 1 2 3 4 5 6 7 8 9 10 11 12 13 \d+ a_0317 Table "public.a_0317" Column | Type | Modifiers | Storage | Stats target | Description --------+---------+-----------+---------+--------------+------------- a | integer | | plain | | Indexes: "a_0317_index" btree (a) LOCAL(PARTITION p1_index, PARTITION p2_inde) TABLESPACE pg_default Range partition by(a) Number of partitions: 2 (View pg_partition to check each partition range.) Has OIDs: no Distribute By: HASH(a) Location Nodes: ALL DATANODES Options: orientation=row, compression=no

问题复现 创建分区表a_0317，含p1，p2两个分区。 1 CREATE TABLE a_0317(a int) partition by range(a) (partition p1 values less than (4), partition p2 values less than (8)); 创建主表与分区索引。 1 CREATE INDEX a_0317_index on a_0317(a) local (partition p1_index, partition p2_inde); 查看分区表分区索引信息如下： 查看主表索引信息。 1 2 3 4 5 6 7 8 9 10 11 SELECT oid,* FROM pg_class where relname ='a_0317_index'; oid | relname | relnamespace | reltype | reloftype | relowner | relam | relfilenode | reltablespace | relpages | reltuples | relallvisible | reltoastrelid | reltoastidxid | reldeltarelid | reld eltaidx | relcudescrelid | relcudescidx | relhasindex | relisshared | relpersistence | relkind | relnatts | relchecks | relhasoids | relhaspkey | relhasrules | relhastriggers | relhassubclass | relcmprs | relhasclusterkey | relrowmovement | parttype | relfrozenxid | relacl | reloptions | relreplident | relfrozenxid64 --------+--------------+--------------+---------+-----------+----------+-------+-------------+---------------+----------+-----------+---------------+---------------+---------------+---------------+----- --------+----------------+--------------+-------------+-------------+----------------+---------+----------+-----------+------------+------------+-------------+----------------+----------------+--------- -+------------------+----------------+----------+--------------+--------+------------+--------------+---------------- 241487 | a_0317_index | 2200 | 0 | 0 | 16393 | 403 | 241487 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | f | f | p | i | 1 | 0 | f | f | f | f | f | 0 | f | f | p | 0 | | | n | 0 (1 row) 根据主表索引信息查看分区索引信息。 1 2 3 4 5 6 7 8 9 10 SELECT * FROM pg_partition where parentid= 241487; relname | parttype | parentid | rangenum | intervalnum | partstrategy | relfilenode | reltablespace | relpages | reltuples | relallvisible | reltoastrelid | reltoastidxid | indextblid | indisusable | reldeltarelid | reldeltaidx | relcudescrelid | relcudescidx | relfrozenxid | intspnum | partkey | intervaltablespace | interval | boundaries | transit | reloptions | relfrozenxid64 ----------+----------+----------+----------+-------------+--------------+-------------+---------------+----------+-----------+---------------+---------------+---------------+------------+-------------+- --------------+-------------+----------------+--------------+--------------+----------+---------+--------------------+----------+------------+---------+------------+---------------- p1_index | x | 241487 | 0 | 0 | n | 241488 | 0 | 0 | 0 | 0 | 0 | 0 | 241485 | t | 0 | 0 | 0 | 0 | 0 | | | | | | | | 0 p2_inde | x | 241487 | 0 | 0 | n | 241489 | 0 | 0 | 0 | 0 | 0 | 0 | 241486 | t | 0 | 0 | 0 | 0 | 0 | | | | | | | | 0 (2 rows) 连接CN开启读写事务，从pg_partition系统表删除p1分区的索引信息。 1 2 START TRANSACTION read write; DELETE from pg_partition where relname = 'p1_index'; 查看表定义报错与现场报错相同，问题复现： 1 2 \d+ a_0317 ERROR: The local index 700633 on the partition 700647 not exist.CONTEXT: referenced column: pg_get_indexdef

使用限制 单账号跟踪的事件可以通过 云审计 控制台查询。多账号的事件只能在账号自己的事件列表页面去查看，或者到组织追踪器配置的OBS桶中查看，也可以到组织追踪器配置的 CTS /system日志流下面去查看。 用户通过云审计控制台只能查询最近7天的操作记录。如果需要查询超过7天的操作记录，您必须配置转储到 对象存储服务 (OBS)或 云日志 服务(LTS)，才可在OBS桶或LTS日志组里面查看历史事件信息。否则，您将无法追溯7天以前的操作记录。 云上操作后，1分钟内可以通过云审计控制台查询管理类事件操作记录，5分钟后才可通过云审计控制台查询数据类事件操作记录。 CTS新版事件列表不显示数据类审计事件，您需要在旧版事件列表查看数据类审计事件。 云审计控制台对用户的操作事件日志保留7天，过期自动删除，不支持人工删除。

获取部门用户基础信息 获取部门用户的基础信息。 输入参数 用户配置获取部门用户基础信息执行动作，相关参数说明如表9所示。 表9 获取部门用户基础信息输入参数说明 参数 必填 说明 分页数 是 分页查询的游标，最开始传0。 分页长度 是 分页长度，最大值100。 部门ID 是 部门ID。 输出参数 用户可以在之后的执行动作中调用该输出参数，输出参数说明请参考表10。 表10 获取部门用户基础信息输出参数说明 参数 说明 errcode 错误码。 errmsg 错误信息。 has_more has_more。 list 部门用户基础信息列表。 name 用户名。 userid 用户ID。 request_id request_id。

创建群会话 在钉钉内创建群会话。 输入参数 用户配置创建群会话执行动作，相关参数说明如表6所示。 表6 创建群会话输入参数说明 参数 必填 说明 access_token 是 access_token。 群名称 否 群名称。 群主 否 群主的userid。 群成员列表 否 群成员的userid数组。 输出参数 用户可以在之后的执行动作中调用该输出参数，输出参数说明请参考表7。 表7 创建群会话输出参数说明 参数 说明 errcode 错误码。 errmsg 错误信息。 chatid 群会话的ID。 openConversationId openConversationId。 conversationTag conversationTag。

订阅顺序消息 只需要在订阅普通消息的代码基础上增加orderly=True，参考如下示例代码（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。 import time from rocketmq.client import PushConsumer, ConsumeStatus def callback(msg): print(msg.id, msg.body, msg.get_property('property')) return ConsumeStatus.CONSUME_SUC CES S def start_consume_message(): consumer = PushConsumer('consumer_group', orderly=True) consumer.set_name_server_address('192.168.0.1:8100') consumer.subscribe('TopicTest', callback) print('start consume message') consumer.start() while True: time.sleep(3600) if __name__ == '__main__': start_consume_message() 示例代码中的参数说明如下，请参考收集连接信息获取参数值。 consumer_group：表示消费组名称。 192.168.0.1:8100：表示实例连接地址和端口。 TopicTest：表示Topic名称。

发送顺序消息 参考如下示例代码（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。 from rocketmq.client import Producer, Message topic = 'TopicTest' gid = 'test' name_srv = '192.168.0.1:8100' def create_message(): msg = Message(topic) msg.set_keys('messageKey') msg.set_tags('messageTag') msg.set_property('property', 'test') msg.set_body('message body') return msg def send_orderly_with_sharding_key(): producer = Producer(gid, True) producer.set_name_server_address(name_srv) producer.start() msg = create_message() ret = producer.send_orderly_with_sharding_key(msg, 'orderId') print('send message status: ' + str(ret.status) + ' msgId: ' + ret.msg_id) producer.shutdown() if __name__ == '__main__': send_orderly_with_sharding_key()

发送事务消息 参考如下示例代码（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。 import time from rocketmq.client import Message, TransactionMQProducer, TransactionStatus topic = 'TopicTest' gid = 'test' name_srv = '192.168.0.1:8100' def create_message(): msg = Message(topic) msg.set_keys('messageKey') msg.set_tags('messageTag') msg.set_property('property', 'test') msg.set_body('message body') return msg def check_callback(msg): print('check: ' + msg.body.decode('utf-8')) return TransactionStatus.COMMIT def local_execute(msg, user_args): print('local: ' + msg.body.decode('utf-8')) return TransactionStatus.UNKNOWN def send_transaction_message(count): producer = TransactionMQProducer(gid, check_callback) producer.set_name_server_address(name_srv) producer.start() for n in range(count): msg = create_message() ret = producer.send_message_in_transaction(msg, local_execute, None) print('send message status: ' + str(ret.status) + ' msgId: ' + ret.msg_id) print('send transaction message done') while True: time.sleep(3600) if __name__ == '__main__': send_transaction_message(10)

生产消息 生产消息的示例代码如下（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。 @SpringBootApplication public class ProduceDemoApplication implements CommandLineRunner { @Resource private RocketMQTemplate rocketMQTemplate; public static void main(String[] args) { SpringApplication.run(ProduceDemoApplication.class, args); } @Override public void run(String... args) throws Exception { // send message synchronously rocketMQTemplate.convertAndSend("topic", "Hello, World!"); // send spring message rocketMQTemplate.send( "topic", MessageBuilder.withPayload("Hello, World! I'm from spring message").build()); // send messgae asynchronously rocketMQTemplate.asyncSend( "topic", MessageBuilder.withPayload("Hello, World! I'm from spring message").build(), new SendCallback() { @Override public void onSuccess(SendResult var1) { System.out.printf("async onSucess SendResult=%s %n", var1); } @Override public void onException(Throwable var1) { System.out.printf("async onException Throwable=%s %n", var1); } }); // Send messages orderly rocketMQTemplate.syncSendOrderly( "topic", MessageBuilder.withPayload("Hello, World").build(), "hashkey"); } }

订阅普通消息 参考如下示例代码（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。 import time from rocketmq.client import PushConsumer, ConsumeStatus def callback(msg): print(msg.id, msg.body, msg.get_property('property')) return ConsumeStatus.CONSUME_SUCCESS def start_consume_message(): consumer = PushConsumer('consumer_group') consumer.set_name_server_address('192.168.0.1:8100') consumer.subscribe('TopicTest', callback) print('start consume message') consumer.start() while True: time.sleep(3600) if __name__ == '__main__': start_consume_message() 示例代码中的参数说明如下，请参考收集连接信息获取参数值。 consumer_group：表示消费组名称。 192.168.0.1:8100：表示实例连接地址和端口。 TopicTest：表示Topic名称。

同步发送 同步发送是指消息发送方发出一条消息到服务端，服务端接收并处理消息，然后返回响应给发送方，发送方收到响应后才会发送下一条消息的通讯方式。 参考如下示例代码（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。 from rocketmq.client import Producer, Message topic = 'TopicTest' gid = 'test' name_srv = '192.168.0.1:8100' def create_message(): msg = Message(topic) msg.set_keys('messageKey') msg.set_tags('messageTag') msg.set_property('property', 'test') msg.set_body('message body') return msg def send_message_sync(): producer = Producer(gid) producer.set_name_server_address(name_srv) producer.start() msg = create_message() ret = producer.send_sync(msg) print('send message status: ' + str(ret.status) + ' msgId: ' + ret.msg_id) producer.shutdown() if __name__ == '__main__': send_message_sync() 示例代码中的参数说明如下，请参考收集连接信息获取参数值。 topic：表示Topic名称。 name_srv：表示实例连接地址和端口。 set_keys：设置消息索引键，可根据关键字精确查找某条消息。 set_tags：设置消息Tag，用于消费端根据指定Tag过滤消息。

概述 收集连接信息介绍如何获取RocketMQ实例连接信息。 Java（TCP协议）~Python（TCP协议）介绍Java、Go和Python客户端访问分布式消息服务RocketMQ版的示例代码，具体如表1所示。 表1 示例代码 客户端语言 示例代码 Java（TCP协议） 收发普通消息 收发顺序消息 收发事务消息 发送定时消息 使用ACL权限访问 消费端限流 Java（gRPC协议） 收发普通消息 收发顺序消息 收发事务消息 发送定时消息 Go（TCP协议） 收发普通消息 收发顺序消息 收发事务消息 发送定时消息 使用ACL权限访问 Go（gRPC协议） 收发普通消息 收发顺序消息 收发事务消息 发送定时消息 Python（TCP协议） 收发普通消息 收发顺序消息 收发事务消息 发送定时消息 使用ACL权限访问

消费者增加用户认证信息 无论是普通消息、顺序消息、定时消息，还是事务消息，都参考如下代码。以下加粗内容需要替换为实例自有信息，请根据实际情况替换。 import time from rocketmq.client import PushConsumer, ConsumeStatus def callback(msg): print(msg.id, msg.body, msg.get_property('property')) return ConsumeStatus.CONSUME_SUCCESS def start_consume_message(): consumer = PushConsumer('consumer_group') consumer.set_name_server_address('192.168.0.1:8100') # 设置权限（角色名和密钥） consumer.set_session_credentials( "ACL_User_Name", # 角色名称 "ACL_Secret_Key", # 角色密钥 '' )#用户名和密钥直接硬编码到代码中或者明文存储都存在很大的风险，建议在配置文件或者环境变量中密文存放，使用时解密。 consumer.subscribe('TopicTest', callback) print('start consume message') consumer.start() while True: time.sleep(3600) if __name__ == '__main__': start_consume_message() 示例代码中的参数说明如下，请参考收集连接信息获取参数值。 consumer_group：表示消费组名称。 192.168.0.1:8100：表示实例连接地址和端口。 ACL_User_Name：表示用户名。创建用户的步骤，请参见创建用户。 ACL_Secret_Key：表示用户的密钥。 TopicTest：表示Topic名称。

生产者增加用户认证信息 普通消息、顺序消息和定时消息，参考如下代码。以下加粗内容需要替换为实例自有信息，请根据实际情况替换。 from rocketmq.client import Producer, Message topic = 'TopicTest' gid = 'test' name_srv = '192.168.0.1:8100' def create_message(): msg = Message(topic) msg.set_keys('messageKey') msg.set_tags('messageTag') msg.set_property('property', 'test') msg.set_body('message body') return msg def send_message_sync(): producer = Producer(gid) producer.set_name_server_address(name_srv) # 设置权限（角色名和密钥） producer.set_session_credentials( "ACL_User_Name", # 角色名称 "ACL_Secret_Key", # 角色密钥 '' )#用户名和密钥直接硬编码到代码中或者明文存储都存在很大的风险，建议在配置文件或者环境变量中密文存放，使用时解密。 producer.start() msg = create_message() ret = producer.send_sync(msg) print('send message status: ' + str(ret.status) + ' msgId: ' + ret.msg_id) producer.shutdown() if __name__ == '__main__': send_message_sync() 示例代码中的参数说明如下，请参考收集连接信息获取参数值。 topic：表示Topic名称。 name_srv：表示实例连接地址和端口。 ACL_User_Name：表示用户名。创建用户的步骤，请参见创建用户。 ACL_Secret_Key：表示用户的密钥。 事务消息，参考如下代码。以下加粗内容需要替换为实例自有信息，请根据实际情况替换。 import time from rocketmq.client import Message, TransactionMQProducer, TransactionStatus topic = 'TopicTest' gid = 'test' name_srv = '192.168.0.1:8100' def create_message(): msg = Message(topic) msg.set_keys('message Key') msg.set_tags('message Tag') msg.set_property('property', 'test') msg.set_body('message body') return msg def check_callback(msg): print('check: ' + msg.body.decode('utf-8')) return TransactionStatus.COMMIT def local_execute(msg, user_args): print('local: ' + msg.body.decode('utf-8')) return TransactionStatus.UNKNOWN def send_transaction_message(count): producer = TransactionMQProducer(gid, check_callback) producer.set_name_server_address(name_srv) # 设置权限（角色名和密钥） producer.set_session_credentials( "ACL_User_Name", # 角色名称 "ACL_Secret_Key", # 角色密钥 '' )#用户名和密钥直接硬编码到代码中或者明文存储都存在很大的风险，建议在配置文件或者环境变量中密文存放，使用时解密。 producer.start() for n in range(count): msg = create_message() ret = producer.send_message_in_transaction(msg, local_execute, None) print('send message status: ' + str(ret.status) + ' msgId: ' + ret.msg_id) print('send transaction message done') while True: time.sleep(3600) if __name__ == '__main__': send_transaction_message(10) 示例代码中的参数说明如下，请参考收集连接信息获取参数值。 topic：表示Topic名称。 name_srv：表示实例连接地址和端口。 set_keys：设置消息索引键，可根据关键字精确查找某条消息。 set_tags：设置消息Tag，用于消费端根据指定Tag过滤消息。 ACL_User_Name：表示用户名。创建用户的步骤，请参见创建用户。 ACL_Secret_Key：表示用户的密钥。

到期后影响 图2描述了包年/包月的数字化制造云平台各个阶段的状态。购买后，在计费周期内正常运行，此阶段为有效期；资源到期而未续费时，将陆续进入宽限期和保留期。 图2 包年/包月的数字化制造云平台生命周期 到期预警 包年/包月的数字化制造云平台在到期前第7天内，系统将向用户推送到期预警消息。预警消息将通过邮件、短信和站内信的方式通知到华为账号的创建者。 到期后影响 当您的包年/包月的数字化制造云平台到期未续费，首先会进入宽限期，宽限期内您可以正常访问数字化制造云平台的对应资源。 如果您在宽限期内仍未续费包年/包月的数字化制造云平台，那么就会进入保留期。保留期内，数据仍予以保留，但是您将无法访问处在保留期的商品，届时将影响使用。 保留期到期后，如果包年/包月的数字化制造云平台仍未续费，对应的数字化制造云平台将终止服务，系统中的数据也将被永久删除且无法恢复。 华为云根据客户等级定义了不同客户的宽限期和保留期时长。 关于续费的详细介绍请参见续费概述。

适用计费项 数字化制造云平台只有同时配置站点和用户，才能确定一个数字化制造云平台的具体服务形态。以下计费项支持包年/包月。 表1 适用计费项 计费项 说明 基础版 数字化制造云平台 站点 随包年/包月的数字化制造云平台配置的站点，部署在公有云上，其计费模式也为包年/包月。 价格随购买站点数量增加而倍数递增。 基础版 数字化制造云平台 用户 随包年/包月的数字化制造云平台配置的用户，部署在公有云上，其计费模式也为包年/包月。 价格随购买用户数量增加而倍数递增。 基础版 数字化制造云平台 边缘站点 随包年/包月的数字化制造云平台配置的边缘站点，部署在边缘云上，其计费模式也为包年/包月。 价格随购买站点数量增加而倍数递增。 基础版 数字化制造云平台 边缘用户 随包年/包月的数字化制造云平台配置的边缘用户，部署在边缘云上，其计费模式也为包年/包月。 价格随购买用户数量增加而倍数递增。 假设您计划购买1个用于部署在公有云上的数字化制造云平台，包含1个站点和100个用户，购买时长为1个月。在购买数字化制造云平台页面底部，您将看到所需费用的明细，如图1所示。 图1 配置费用示例 配置费用将包括以下部分： 站点：根据所选服务对应的站点数量计算的费用。 用户：根据所选服务对应的用户数量计算的费用。

使用限制 随数字化制造云平台购买时所配置的站点和用户资源不支持转移，后续变更规格时仅支持资源升配，不支持资源降配。即变更规格时，已购站点的用户数量可以增加，不可以减少。 此处以部署在公有云上，某企业购买了2个数字化制造云平台为例，分别命名为制造测试服务和制造生产服务。其中，制造测试服务在购买时配置了1个站点和50个用户，制造生产服务配置了2个站点和200个用户。在使用过程中，用户发现制造测试服务的规格无法满足业务需要。此时，仅支持企业通过变更规格进行资源升配，增加制造测试服务中的站点和用户数量。但不支持将制造生产服务中的站点和用户资源转移到制造测试服务。

变更配置后对计费的影响 当前包年/包月的数字化制造云平台规格不满足您的业务需要时，您可以在CraftArts IPDCenter控制台发起变更规格操作。包年/包月的数字化制造云平台仅支持资源升配，不支持资源降配。 资源升配时，新配置价格高于旧配置价格，此时您需要支付新旧配置的差价（即升配费用）。以无任何优惠的场景为例，计算公式如下： 升配费用 =（新配置价格 - 旧配置价格）* 剩余周期 其中： 包月剩余周期为：每个自然月的剩余天数/对应自然月的最大天数，变更当天不算。 包年剩余周期为：剩余天数（不含闰年的2.29号）/365，变更当天不算。 下面以资源升配且无任何优惠的场景为例，计算价格仅作示例用，非官网实际价格，实际价格请参考硬件开发工具链平台云服务价格详情。 假设您在2024/03/08购买了1个包年/包月的用于部署在公有云上的数字化制造云平台，包含1个站点和100个用户，购买时长为1个月；在2024/03/18变更规格为1个站点和200个用户。旧配置价格为35,000元/月，新配置价格为50,000元/月，则变更规格时您需要支付的升配费用为10,290元。计算公式如下： 升配费用 =（50,000 - 35,000）* （13 / 31 + 8 / 30）= 10,290（元） 其中，13表示3月份剩余天数，31表示3月份最大天数；8表示4月份剩余天数，30表示4月份最大天数。 更多信息请参见变更资源规格费用说明。

计费场景-包年/包月的数字化制造云平台 某用户于2024/03/08 15:30:00购买了1个包年/包月的数字化制造云平台（MBM Space），计费时长为1个月，用于部署在公有云上，购买规格配置如下： 基础版 数字化制造云平台 站点：1个 基础版 数字化制造云平台 用户：100人 用了一段时间后，用户发现该服务当前规格无法满足业务需要，于2024/03/18 9:00:00升级规格为： 基础版 数字化制造云平台 站点：1个 基础版 数字化制造云平台 用户：200人 那么在3~4月份，该服务总共需要支付多少费用呢？

计费构成分析-包年/包月的数字化制造云平台 以下计算价格仅作示例用，非官网实际价格，实际价格请参考硬件开发工具链平台云服务价格详情。 可以将服务的使用阶段按照购买次数分为两段：2024/03/08 15:30:00 ~ 2024/04/08 23:59:59期间为初次购买的包年/包月计费，2024/03/18 9:00:00 ~ 2024/04/08 23:59:59期间二次购买，计升配费用，计费时长为剩余周期。 在2024/03/08 15:30:00 ~ 2023/04/08 23:59:59期间初次购买，为包年/包月计费，计费时长为1个月。费用计算如下： 在2024/03/18 9:00:00 ~ 2024/04/08 23:59:59期间二次购买，计升配费用，计费时长为剩余周期。费用计算如下： 根据资源升配且无任何优惠场景的计费公式，可知： 升配费用 =（50,000 - 35,000）*（13 / 31 + 8 / 30）= 10,290（元） 其中，13表示3月份剩余天数，31表示3月份最大天数；8表示4月份剩余天数，30表示4月份最大天数。 由此可见，在3~4月份，该服务总共产生的费用为：35,000 + 10,290 = 45,290（元）。

计费项 工业仿真云平台服务不同的部署位置有不同的计费项，具体内容如表1所示。 表1 工业仿真云平台服务计费项 部署位置 计费项 计费项说明 适用的计费模式 计费公式 公有云 仿真管理服务-基础版 仿真管理服务-基础版按购买的用户数进行计费。用户表示使用工业仿真云平台服务的最终用户，组织分配的使用用户数需小于组织购买的用户数。 计费因子：用户数 包年/包月 仿真管理服务-基础版单价* 购买用户数*购买时长 仿真管理服务-基础版单价以售卖页显示的价格为准。 仿真计算服务 不同规格的计算实例和不同的节点数量，能提供不同的计算和存储能力，您可根据实际需要选择合适的实例类型和节点数。 计费因子：实例类型、节点数 包年/包月 计算实例规格单价* 购买节点数*购买时长 仿真计算实例规格单价以售卖页显示的价格为准。 边缘云 仿真管理服务-专业版-边缘 仿真管理服务-专业版按购买的用户数进行计费。用户表示使用工业仿真云平台服务的最终用户，组织分配的使用用户数需小于组织购买的用户数。 计费因子：用户数 包年/包月 仿真管理服务-专业版-边缘单价* 购买用户数*购买时长 仿真管理服务-专业版-边缘单价以售卖页显示的价格为准。 IPDCenter基础服务-边缘 仿真管理服务-专业版依赖IPDCenter基础服务，在开通仿真管理服务-专业版时，会同时开通IPDCenter基础服务。 计费因子：用户数 包年/包月 IPDCenter基础服务 - 边缘单价* 购买用户数*购买时长 IPDCenter基础服务-边缘单价以售卖页显示的价格为准。 父主题： 工业仿真云平台服务计费说明