如下ahttp的搜索结果-阿里云开发者社区

"如下ahttp" 检索

共 59802 条结果

数据湖存储的安全写入之道

作者：焱冰@阿里云焱冰背景数据湖的兴起，给数据存储带来了一轮新的革命。越来越多的公司选择将存储切换到云上对象存储。因为云上对象存储往往意味着大容量、低成本、易扩容。说到对象存储，必然涉及到 S3 协议，S3 协议已经事实上成为对象存储的通用协议。不过，市面上不少数据平台公司，也会选择基于 S3 协议又兼顾 Hadoop 使用习惯的 S3A Connector，比如 Databricks 在对象存储上提供的表数据结构 Delta Lake。我们就以 Hadoop 社区中的 S3A Connector 的实现为切入，来分析一下数据湖写入路径的安全性。Hadoop S3 的写入支持因为 S3 协议本身不支持增量写入，因此 S3A 实现时默认的写入方式是先通过缓存到本地，最后在文件 close 后再上传到对象存储。但是，这种默认的方式并不一定高效，对大文件来说，在 close 被调用前，本地已经缓存大量的数据，这样会造成 close 操作非常耗时，文件写入整体看也不高效。从 Hadoop 2.8.5 版本开始，可以通过设置 fs.s3a.fast.upload 为 true，打开快速上传来优化写入路径。打开后，可以边写本地缓存块，边将满足大小的块异步上传（默认 100M 一个块）。这样也满足了对象存储中分阶段上传接口的一些限制，比如单个块不能小于 5M，分块总数不能大于 10000。通过阅读 Hadoop 2.8.5 相关源码，我们可以发现打开 fs.s3a.fast.upload 后，S3AFileSystem 在创建文件时会打开 S3ABlockOutputStream（Hadoop 3.x 也有类似的 S3AFastOutputStream）随后，S3ABlockOutputStream 在处理 write、flush 等操作时，则会调用一个抽象的 S3ADataBlock 来执行。而 S3ADataBlock 则可由三种工厂方法来创建，分别创建基于堆内存的 ArrayBlock、基于磁盘的 DiskBlock，或者基于堆外内存的 ByteBufferBlock。选择哪种工厂，由 fs.s3a.fast.upload.buffer 这个配置项控制，默认为磁盘（disk）。其他两种可选配置为堆内存（array）和堆外内存（bytebuffer）。磁盘的问题通过了解 Hadoop 社区中 S3A 的实现，我们发现借助磁盘缓存数据是常见甚至默认的行为。因为这样可以减少内存占用，缓存更多的数据。但是，这样也带来了磁盘本身的阿喀琉斯之踵 -- 磁盘的稳定性问题。在数据存储领域，磁盘的问题往往非常令人头疼。比如磁盘写满，磁盘坏道问题，还有偶现的磁盘数据比特反转导致的数据安全性问题。哪怕单块磁盘的可靠性非常高，但由于磁盘出现问题的概率会随着磁盘数的提升而变大，这会使数据安全性蒙上一层阴影。对于R个副本的情况，设磁盘的年故障率为P，磁盘数为N，则整个机群有C ( N, R ) = N! / ( R! * ( N- R )! ) 种 R 副本的组合方式。机群数据总量为 M，分片大小为 T，那么有 R 个磁盘同时损坏造成数据丢失的概率是：* 引用于《磁盘故障与存储系统的年失效率估算》因此，要保证写路径的数据安全型，我们不能完全依赖底层存储介质的保证。仍需要我们在数据写入时就做一些努力。我们先来做一些实验来看看 S3AFileSystem 在这些问题上的表现。模拟磁盘 IO 问题修改 core-sites.xml 中的 fs.s3a.buffer.dir 指向 /dev/vdc 所在的路径，比如我机器上的 /data2/ <property> <name>fs.s3a.fast.upload</name> <value>true</value> </property> <property>  <name>fs.s3a.buffer.dir</name> <value>/data2/tmp/</value> </property>创建并运行 stap 脚本，对所有在 /dev/vdc 上写操作的返回 IO Error#!/usr/bin/stap probe vfs.write.return { if (devname == "vdc") { $return = -5 } } $ stap -g io_errno.stp执行写入程序 demo，验证 stap 脚本有效$ dd if=/dev/zero of=test-1G-stap bs=1G count=1 $ hadoop fs -put test-1G s3a://<your-bucket>/返回结果：put: 输入/输出错误可以发现相关操作能正确抛出 IO 错误。模拟磁盘比特反转魔改 libfuse passthrough 中的 write 方法，并将 /data2/ 通过 fuse 挂载到 /mnt/passthrough$ mkdir -p /mnt/passthrough/ $ ./passthrough /mnt/passthrough/ -omodules=subdir -osubdir=/data2/ -oauto_unmount修改 core-sites.xml 中的 hadoop.tmp.dir 指向 /mnt/passthrough<property> <name>fs.s3a.fast.upload</name> <value>true</value> </property> <property>  <name>fs.s3a.buffer.dir</name> <value>/mnt/passthrough/</value> </property>执行写入程序 demo，验证上传内容的正确性。$ mkdir -p input output $ dd if=/dev/zero of=input/test-1G-fuse bs=1G count=1 $ hadoop fs -put input/test-1G-fuse s3a://<your-bucket>/ $ hadoop fs -get s3a://<your-bucket>/test-1G-fuse output/ $ md5sum input/test-1G-fuse output/test-1G-fuse返回结果：cd573cfaace07e7949bc0c46028904ff input/test-1G-fuse 37eb6e664e706ea48281acbd4676569e output/test-1G-fuse可以发现，输入和输出的数据并不一致。综上，通过 Hadoop S3AFileSystem 写入可以发现磁盘 IO 问题并正确抛出异常，但无法发现磁盘比特反转问题。网络的问题既然磁盘写入有问题，那我们使用内存写入是否就一定可以避免踩坑呢？答案是不能，还可能有网络问题。Amazon S3 在 2008 年就曾因为网络问题导致的比特位反转引发过重大事故。后来，大家分析这种问题多发生于两端间隔多个路由器的情况，路由器可能因为硬件/内存故障导致单/多比特位反转或双字节交换，这种反转如果发生在 payload 区，则无法通过链路层、网络层、传输层的 checksum 检查出来。因此 Amazon S3 在这次事故中吸取的教训是，要通过在应用层给所有东西都添加 checksum 来保证数据正确性。让我们来做一个实验，来看看 S3 是怎么做到 Checksum all of the things的，又是否能防止网络比特反转或者网络丢包呢？模拟网络比特反转安装 mitmproxy$ pip3 install mitmproxy $ mitmproxy --version Mitmproxy: 5.3.0 Python: 3.6.8 OpenSSL: OpenSSL 1.1.1h 22 Sep 2020 Platform: Linux-3.10.0-1160.71.1.el7.x86_64-x86_64-with-centos-7.9.2009-Core利用 mitmdump 反向代理 s3a endpoint，并篡改其中的写请求。编写 addons.pyfrom mitmproxy import ctx, http import json import time import os class HookOssRequest: def request(self, flow: http.HTTPFlow): print("") print("="*50) print("FOR: " + flow.request.url) print(flow.request.method + " " + flow.request.path + " " + flow.request.http_version) print("-"*50 + "request headers:") for k, v in flow.request.headers.items(): print("%-20s: %s" % (k.upper(), v)) if flow.request.host == "<your-bucket>.oss-cn-shanghai-internal.aliyuncs.com" and flow.request.method == "PUT": clen = len(flow.request.content) rbit = ord('a') clist = list(flow.request.content) origin = clist[clen - 1] clist[clen - 1] = rbit updated = clist[clen - 1] flow.request.content = bytes(clist) ctx.log.info("updated requesting content pos(" + str(clen - 1) + ") from " + str(chr(origin)) + " to " + str(chr(updated))) def response(self, flow: http.HTTPFlow): pass addons = [ HookOssRequest() ]反向代理 http://.oss-cn-shanghai-internal.aliyuncs.com 到 http://localhost:8765$ mitmdump -s addons.py -p 8765 --set block_global=false --mode reverse:http://<your-bucket>.oss-cn-shanghai-internal.aliyuncs.com修改 core-sites.xml 中的 fs.s3a.endpoint 指向 localhost:8765，并关闭ssl。<property> <name>fs.s3a.connection.ssl.enabled</name> <value>false</value> </property> <property> <name>fs.s3a.fast.upload</name> <value>true</value> </property>执行写入程序 demo，验证上传内容的正确性$ mkdir -p input output $ dd if=/dev/zero of=input/test-100M-proxy bs=$(( 100*1024*1024 + 1 )) count=1 $ hadoop fs -put input/test-100M-proxy s3a://<your-bucket>/返回结果：xx/xx/xx xx:xx:xx WARN s3a.S3ABlockOutputStream: Transfer failure of block FileBlock{index=2, destFile=/data/hadoop/hadoop-2.8.5/tmp/s3a/s3ablock-0002-6832685202941984333.tmp, state=Upload, dataSize=1, limit=104857600} xx/xx/xx xx:xx:xx WARN s3a.S3ABlockOutputStream: Transfer failure of block FileBlock{index=1, destFile=/data/hadoop/hadoop-2.8.5/tmp/s3a/s3ablock-0001-635596269039598032.tmp, state=Closed, dataSize=104857600, limit=104857600} put: Multi-part upload with id '14ABE04E57114D0D9D8DBCFE4CB9366E' to test-100M-proxy._COPYING_ on test-100M-proxy._COPYING_: com.amazonaws.AmazonClientException: Unable to verify integrity of data upload. Client calculated content hash (contentMD5: 93B885ADFE0DA089CDF634904FD59F71 in hex) didn't match hash (etag: 0CC175B9C0F1B6A831C399E269772661 in hex) calculated by Amazon S3. You may need to delete the data stored in Amazon S3. (bucketName: <your-bucket>, key: test-100M-proxy._COPYING_, uploadId: 14ABE04E57114D0D9D8DBCFE4CB9366E, partNumber: 2, partSize: 1): Unable to verify integrity of data upload. Client calculated content hash (contentMD5: 93B885ADFE0DA089CDF634904FD59F71 in hex) didn't match hash (etag: 0CC175B9C0F1B6A831C399E269772661 in hex) calculated by Amazon S3. You may need to delete the data stored in Amazon S3. (bucketName: <your-bucket>, key: test-100M-proxy._COPYING_, uploadId: 14ABE04E57114D0D9D8DBCFE4CB9366E, partNumber: 2, partSize: 1)可见，Amazon S3 在 header 签名中强制对每个 upload part 的 payload 做了 Content-MD5 的校验，能够有效检测出网络比特反转。模拟网络丢包之前的测试验证了， S3 使用 Content-MD5 的校验可以保证单个请求的正确性，但在写一些大文件，或是涉及 JobCommitter 的作业中，往往会使用 multipart upload 来进行并发上传。而网络丢包也是一种常见的问题。于是，接下来我们来验证下，如果上传过程中其中一个 part 丢失，是否会给上传结果造成影响。同样使用 mitmproxy 来模拟丢包利用 mitmdump 反向代理 s3a endpoint，并丢弃其中 part2 的请求。编写 addons.pyfrom mitmproxy import ctx, http import json import time import os class HookOssRequest: def request(self, flow: http.HTTPFlow): print("") print("="*50) print("FOR: " + flow.request.url) print(flow.request.method + " " + flow.request.path + " " + flow.request.http_version) print("-"*50 + "request headers:") for k, v in flow.request.headers.items(): print("%-20s: %s" % (k.upper(), v)) if flow.request.host == "<your-bucket>.oss-cn-shanghai-internal.aliyuncs.com" and flow.request.method == "PUT": if "partNumber=2" in flow.request.path: flow.response = http.HTTPResponse.make( 200, # (optional) status code b"Hello World", # (optional) content {"Content-Type": "text/html"}, # (optional) headers ) ctx.log.info("drop part-2 request!") ctx.log.info("requesting length:" + str(len(flow.request.content))) def response(self, flow: http.HTTPFlow): pass addons = [ HookOssRequest() ]反向代理 http://.oss-cn-shanghai-internal.aliyuncs.com 到 http://localhost:8765$ mitmdump -s addons.py -p 8765 --set block_global=false --mode reverse:http://<your-bucket>.oss-cn-shanghai-internal.aliyuncs.com同样修改 core-sites.xml 中的 fs.s3a.endpoint 指向 localhost:8765，并关闭ssl。<property> <name>fs.s3a.connection.ssl.enabled</name> <value>false</value> </property> <property> <name>fs.s3a.fast.upload</name> <value>true</value> </property>执行写入程序 demo，验证上传内容的正确性$ mkdir -p input output $ dd if=/dev/zero of=input/test-100M-proxy bs=$(( 100*1024*1024 + 1 )) count=1 $ hadoop fs -put input/test-100M-proxy s3a://<your-bucket>/ xx/xx/xx xx:xx:x WARN s3a.S3ABlockOutputStream: Transfer failure of block FileBlock{index=2, destFile=/data/hadoop/hadoop-2.8.5/tmp/s3a/s3ablock-0002-2063629354855241099.tmp, state=Upload, dataSize=1, limit=104857600} put: Multi-part upload with id 'D58303E74A5F4E6D8A27DD112297D0BE' to test-100M-proxy._COPYING_ on test-100M-proxy._COPYING_: com.amazonaws.AmazonClientException: Unable to verify integrity of data upload. Client calculated content hash (contentMD5: 93B885ADFE0DA089CDF634904FD59F71 in hex) didn't match hash (etag: null in hex) calculated by Amazon S3. You may need to delete the data stored in Amazon S3. (bucketName: <your-bucket>, key: test-100M-proxy._COPYING_, uploadId: D58303E74A5F4E6D8A27DD112297D0BE, partNumber: 2, partSize: 1): Unable to verify integrity of data upload. Client calculated content hash (contentMD5: 93B885ADFE0DA089CDF634904FD59F71 in hex) didn't match hash (etag: null in hex) calculated by Amazon S3. You may need to delete the data stored in Amazon S3. (bucketName: <your-bucket>, key: test-100M-proxy._COPYING_, uploadId: D58303E74A5F4E6D8A27DD112297D0BE, partNumber: 2, partSize: 1)可见，Amazon S3 在 close 请求中通过 CompleteMultipartUpload 对每个上传的 Part 做了检查，能够发现丢失的请求。校验算法的选择上文已经证明了校验码的不可或缺性，而且可以看到 Amazon S3 默认采用了 MD5 作为校验码。那就是最优的选择了吗？让我们来看看还有没有别的选择。数据摘要算法MD5、SHA-1、SHA-256、SHA-512都是数据摘要算法，均被广泛作为密码的散列函数。但由于MD5、SHA-1已经被证明为不安全的算法，目前建议使用较新的SHA-256和SHA-512。所有算法的输入均可以是不定长的数据。MD5输出是16字节（128位），SHA-1输出为20字节（160位），SHA-256为32字节（256位），SHA-512为64字节（512位）。可以看到，SHA算法的输出长度更长，因此更难发生碰撞，数据也更为安全。但运算速度与MD5相比，也更慢。循环冗余校验循环冗余校验又称 CRC（Cyclic redundancy check），将待发送的比特串看做是系数为 0 或者 1 的多项式。M = 1001010M(x) = 1*x^6 + 0*x^5 + 0*x^4 + 1*x^3 + 0*x^2 + 1*x^1 + 0*x^0M(x) = x^6 + x^3 + xCRC 编码时，发送方和接收方必须预先商定一个生成多项式 G(x)。发送方将比特串和生成多项式 G(x) 进行运算得到校验码，在比特串尾附加校验码，使得带校验码的比特串的多项式能被 G(x) 整除。接收方接收到后，除以 G(x)，若有余数，则传输有错。校验算法的开销CRC算法的优点是算法实现相对简单、运算速度较快。而且错误检错能力很强，因此被广泛应用于通信数据校验。我们做了一些简单的benchmark以供参考：CRC32 > CRC64 > MD5 > SHA-1 > SHA-512 > SHA-256校验算法单次操作耗时BenchmarkMD5_100MB-8175423280 ns/opBenchmarkSHA1_100MB-8176478051 ns/opBenchmarkSHA256_100MB-8344191216 ns/opBenchmarkSHA512_100MB-8226938072 ns/opBenchmarkCRC32IEEE_100MB-810500107 ns/opBenchmarkCRC32Castagnoli_100MB-812991050 ns/opBenchmarkCRC64_100MB-8 86377178 ns/op而 OSS 支持的校验算法有 MD5 和 CRC64，那么同样的场景下，我们会优先选择 CRC64 替代 MD5。阿里云EMR JindoSDK 的最佳实践在总结了 S3AFileSystem 做法中的优缺点，并结合 OSS 自身提供的一些功能取长补短后，阿里云EMR JindoSDK 得出了自己的最佳实践。JindoSDK 实现的 JindoOutputStream 支持了两种校验方式，一种是请求级别的校验，一种是文件块级别的校验。请求级别的校验，默认关闭。需要打开时，配置 fs.oss.checksum.md5.enable 为 true 即可。配置好之后，客户端会在块级别的请求（PutObject/MultipartUpload）Header 中添加 Payload 的 Content-MD5。如果服务端计算 Payload 的 md5 与客户端提供的不符，则客户端会重试。文件块级别的校验，默认打开。需要关闭时，需要配置 fs.oss.checksum.crc64.enable 为 false。则是在写入流一开始就在内存中同步计算传入 Buffer 的 CRC64，并在文件块落盘时和服务端计算返回的 CRC64 进行比较。使用最新的 jindosdk-4.6.2 版本与 S3AFileSystem 在数据湖写入路径上，综合对比的结果如下：场景S3AFileSystemJindoOssFileSystem磁盘 IO 问题抛出异常java.io.IOException抛出异常java.io.IOException磁盘比特反转未抛出异常抛出异常java.io.IOException网络比特反转抛出异常org.apache.hadoop.fs.s3a.AWSClientIOException抛出异常java.io.IOException 网络丢包抛出异常org.apache.hadoop.fs.s3a.AWSClientIOException抛出异常java.io.IOException写一个 5G 文件的耗时13.375s6.849s可以看到 EMR JindoSDK 在写 OSS 时，不仅有着相比 S3AFileSystem 更完善的错误检查，性能也更为优异。总结与展望数据湖存储的安全写入，必须要能考虑到内存、磁盘、网络的不可靠性。同时，也要结合存储介质本身的特性，选择合适的校验算法。熟悉数据写入完整链路，全面地考虑各种可能遇到的问题，并提供完善的测试方案验证可行性，才算有始有终。阿里云EMR JindoSDK 通过以上方式形成了自己的最佳实践，不仅保证了对象存储写入链路的安全性，同样也支持了EMR JindoFS服务（OSS-HDFS）的写入链路。虽然 OSS-HDFS 中的一个文件可以对应 OSS 上的多个对象，但是在写入 OSS 时，底层复用了同一套实现。因此，在使用时也不需要做额外的适配，完全可以共用相同的配置项。未来我们还将结合 OSS-HDFS，提供在数据随机读场景的安全性校验，而这是对象存储本身目前无法做到的。附录一：测试 S3A 的配置方式core-sites.xml<property> <name>fs.s3a.impl</name> <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value> </property> <property> <name>fs.AbstractFileSystem.s3a.impl</name> <value>org.apache.hadoop.fs.s3a.S3A</value> </property> <property> <name>fs.s3a.access.key</name> <value>xxx</value> </property> <property> <name>fs.s3a.secret.key</name> <value>xx</value> </property> <property> <name>fs.s3a.endpoint</name> <value>localhost:8765</value> </property> <property> <name>fs.s3a.connection.ssl.enabled</name> <value>false</value> </property> <property> <name>fs.s3a.fast.upload</name> <value>true</value> </property> <property>  <name>fs.s3a.buffer.dir</name> <value>/mnt/passthrough/</value> </property>附录二：测试EMR JindoSDK 的配置方式core-sites.xml<property> <name>fs.AbstractFileSystem.oss.impl</name> <value>com.aliyun.jindodata.oss.OSS</value> </property> <property> <name>fs.oss.impl</name> <value>com.aliyun.jindodata.oss.JindoOssFileSystem</value> </property> <property> <name>fs.oss.accessKeyId</name> <value>xxx</value> </property> <property> <name>fs.oss.accessKeySecret</name> <value>xxx</value> </property> <property> <name>fs.oss.endpoint</name>  <value>oss-cn-xxx.aliyuncs.com</value> </property> <property>  <name>fs.oss.tmp.data.dirs</name> <value>/data2/tmp/</value> </property> <property>  <name>fs.oss.second.level.domain.enable</name> <value>true</value> </property>log4j.propertieslog4j.logger.com.aliyun.jindodata=INFO log4j.logger.com.aliyun.jindodata.common.FsStats=INFOmitmproxy获取 endpoint ipping oss-cn-shanghai-internal.aliyuncs.com 64 bytes from xxx.xxx.xxx.xx (xxx.xxx.xxx.xx): icmp_seq=1 ttl=102 time=0.937 ms将 addons.py 中使用 ip 代替 .oss-cn-shanghai-internal.aliyuncs.com if flow.request.host == "xxx.xxx.xxx.xx" and flow.request.method == "PUT":反向代理时也使用 ip 代替 .oss-cn-shanghai-internal.aliyuncs.commitmdump -s addons.py -p 8765 --set block_global=false --mode reverse:http://xxx.xxx.xxx.xx:80欢迎感兴趣的朋友加入钉钉交流群（钉钉搜索群号33413498 或钉钉扫描下方二维码）

文章

存储 · 缓存 · 分布式计算 · 算法 · 安全 · Hadoop · 网络安全 · 对象存储 · 数据安全/隐私保护 · 网络架构

2023-03-09

高德Go生态的服务稳定性建设｜性能优化的实战总结

本文共同作者：阳迪、联想、君清前言go语言凭借着优秀的性能，简洁的编码风格，极易使用的协程等优点，逐渐在各大互联网公司中流行起来。而高德业务使用go语言已经有3年时间了，随着高德业务的发展，go语言生态也日趋完善，今后会有越来越多新的go服务出现。在任何时候，保障服务的稳定性都是首要的，go服务也不例外，而性能优化作为保障服务稳定性，降本增效的重要手段之一，在高德go服务日益普及的当下显得愈发重要。此时此刻，我们将过去go服务开发中的性能调优经验进行总结和沉淀，为您呈上这篇精心准备的go性能调优指南。通过本文您将收获以下内容： 从理论的角度，和你一起捋清性能优化的思路，制定最合适的优化方案。推荐几款go语言性能分析利器，与你一起在性能优化的路上披荆斩棘。总结归纳了众多go语言中常用的性能优化小技巧，总有一个你能用上。基于高德go服务百万级QPS实践，分享几个性能优化实战案例，让性能优化不再是纸上谈兵。1. 性能调优-理论篇1.1 衡量指标优化的第一步是先衡量一个应用性能的好坏，性能良好的应用自然不必费心优化，性能较差的应用，则需要从多个方面来考察，找到木桶里的短板，才能对症下药。那么如何衡量一个应用的性能好坏呢？最主要的还是通过观察应用对核心资源的占用情况以及应用的稳定性指标来衡量。所谓核心资源，就是相对稀缺的，并且可能会导致应用无法正常运行的资源，常见的核心资源如下：cpu：对于偏计算型的应用，cpu往往是影响性能好坏的关键，如果代码中存在无限循环，或是频繁的线程上下文切换，亦或是糟糕的垃圾回收策略，都将导致cpu被大量占用，使得应用程序无法获取到足够的cpu资源，从而响应缓慢，性能变差。内存：内存的读写速度非常快，往往不是性能的瓶颈，但是内存相对来说容量有限切价格昂贵，如果应用大量分配内存而不及时回收，就会造成内存溢出或泄漏，应用无法分配新的内存，便无法正常运行，这将导致很严重的事故。带宽：对于偏网络I/O型的应用，例如网关服务，带宽的大小也决定了应用的性能好坏，如果带宽太小，当系统遇到大量并发请求时，带宽不够用，网络延迟就会变高，这个虽然对服务端可能无感知，但是对客户端则是影响甚大。磁盘：相对内存来说，磁盘价格低廉，容量很大，但是读写速度较慢，如果应用频繁的进行磁盘I/O，那性能可想而知也不会太好。以上这些都是系统资源层面用于衡量性能的指标，除此之外还有应用本身的稳定性指标：异常率：也叫错误率，一般分两种，执行超时和应用panic。panic会导致应用不可用，虽然服务通常都会配置相应的重启机制，确保偶然的应用挂掉后能重启再次提供服务，但是经常性的panic，会导致应用频繁的重启，减少了应用正常提供服务的时间，整体性能也就变差了。异常率是非常重要的指标，服务的稳定和可用是一切的前提，如果服务都不可用了，还谈何性能优化。响应时间(RT)：包括平均响应时间，百分位(top percentile)响应时间。响应时间是指应用从收到请求到返回结果后的耗时，反应的是应用处理请求的快慢。通常平均响应时间无法反应服务的整体响应情况，响应慢的请求会被响应快的请求平均掉，而响应慢的请求往往会给用户带来糟糕的体验，即所谓的长尾请求，所以我们需要百分位响应时间，例如tp99响应时间，即99%的请求都会在这个时间内返回。吞吐量：主要指应用在一定时间内处理请求/事务的数量，反应的是应用的负载能力。我们当然希望在应用稳定的情况下，能承接的流量越大越好，主要指标包括QPS(每秒处理请求数)和QPM(每分钟处理请求数)。1.2 制定优化方案明确了性能指标以后，我们就可以评估一个应用的性能好坏，同时也能发现其中的短板并对其进行优化。但是做性能优化，有几个点需要提前注意：第一，不要反向优化。比如我们的应用整体占用内存资源较少，但是rt偏高，那我们就针对rt做优化，优化完后，rt下降了30%，但是cpu使用率上升了50%，导致一台机器负载能力下降30%，这便是反向优化。性能优化要从整体考虑，尽量在优化一个方面时，不影响其他方面，或是其他方面略微下降。第二，不要过度优化。如果应用性能已经很好了，优化的空间很小，比如rt的tp99在2ms内，继续尝试优化可能投入产出比就很低了，不如将这些精力放在其他需要优化的地方上。由此可见，在优化之前，明确想要优化的指标，并制定合理的优化方案是很重要的。常见的优化方案有以下几种：优化代码有经验的程序员在编写代码时，会时刻注意减少代码中不必要的性能消耗，比如使用strconv而不是fmt.Sprint进行数字到字符串的转化，在初始化map或slice时指定合理的容量以减少内存分配等。良好的编程习惯不仅能使应用性能良好，同时也能减少故障发生的几率。总结下来，常用的代码优化方向有以下几种：提高复用性，将通用的代码抽象出来，减少重复开发。池化，对象可以池化，减少内存分配；协程可以池化，避免无限制创建协程打满内存。并行化，在合理创建协程数量的前提下，把互不依赖的部分并行处理，减少整体的耗时。异步化，把不需要关心实时结果的请求，用异步的方式处理，不用一直等待结果返回。算法优化，使用时间复杂度更低的算法。使用设计模式设计模式是对代码组织形式的抽象和总结，代码的结构对应用的性能有着重要的影响，结构清晰，层次分明的代码不仅可读性好，扩展性高，还能避免许多潜在的性能问题，帮助开发人员快速找到性能瓶颈，进行专项优化，为服务的稳定性提供保障。常见的对性能有所提升的设计模式例如单例模式，我们可以在应用启动时将需要的外部依赖服务用单例模式先初始化，避免创建太多重复的连接。空间换时间或时间换空间在优化的前期，可能一个小的优化就能达到很好的效果。但是优化的尽头，往往要面临抉择，鱼和熊掌不可兼得。性能优秀的应用往往是多项资源的综合利用最优。为了达到综合平衡，在某些场景下，就需要做出一些调整和牺牲，常用的方法就是空间换时间或时间换空间。比如在响应时间优先的场景下，把需要耗费大量计算时间或是网络i/o时间的中间结果缓存起来，以提升后续相似请求的响应速度，便是空间换时间的一种体现。使用更好的三方库在我们的应用中往往会用到很多开源的第三方库，目前在github上的go开源项目就有173万+。有很多go官方库的性能表现并不佳，比如go官方的日志库性能就一般，下面是zap发布的基准测试信息（记录一条消息和10个字段的性能表现）。PackageTimeTime % to zapObjects Allocated⚡️ zap862 ns/op+0%5 allocs/op⚡️ zap (sugared)1250 ns/op+45%11 allocs/opzerolog4021 ns/op+366%76 allocs/opgo-kit4542 ns/op+427%105 allocs/opapex/log26785 ns/op+3007%115 allocs/oplogrus29501 ns/op+3322%125 allocs/oplog1529906 ns/op+3369%122 allocs/op从上面可以看出zap的性能比同类结构化日志包更好，也比标准库更快，那我们就可以选择更好的三方库。2. 性能调优-工具篇当我们找到应用的性能短板，并针对短板制定相应优化方案，最后按照方案对代码进行优化之后，我们怎么知道优化是有效的呢？直接将代码上线，观察性能指标的变化，风险太大了。此时我们需要有好用的性能分析工具，帮助我们检验优化的效果，下面将为大家介绍几款go语言中性能分析的利器。2.1 benchmarkGo语言标准库内置的 testing 测试框架提供了基准测试(benchmark)的能力，benchmark可以帮助我们评估代码的性能表现，主要方式是通过在一定时间(默认1秒)内重复运行测试代码，然后输出执行次数和内存分配结果。下面我们用一个简单的例子来验证一下，strconv是否真的比fmt.Sprint快。首先我们来编写一段基准测试的代码，如下：package main import ( "fmt" "strconv" "testing" ) func BenchmarkStrconv(b *testing.B) { for n := 0; n < b.N; n++ { strconv.Itoa(n) } } func BenchmarkFmtSprint(b *testing.B) { for n := 0; n < b.N; n++ { fmt.Sprint(n) } }我们可以用命令行go test -bench . 来运行基准测试，输出结果如下：goos: darwin goarch: amd64 pkg: main cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkStrconv-12 41988014 27.41 ns/op BenchmarkFmtSprint-12 13738172 81.19 ns/op ok main 7.039s可以看到strconv每次执行只用了27.41纳秒，而fmt.Sprint则是81.19纳秒，strconv的性能是fmt.Sprint的三倍，那为什么strconv要更快呢？会不会是这次运行时间太短呢？为了公平起见，我们决定让他们再比赛一轮，这次我们延长比赛时间，看看结果如何。通过go test -bench . -benchtime=5s 命令，我们可以把测试时间延长到5秒，结果如下:goos: darwin goarch: amd64 pkg: main cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkStrconv-12 211533207 31.60 ns/op BenchmarkFmtSprint-12 69481287 89.58 ns/op PASS ok main 18.891s结果有些变化，strconv每次执行的时间上涨了4ns，但变化不大，差距仍有2.9倍。但是我们仍然不死心，我们决定让他们一次跑三轮，每轮5秒，三局两胜。通过go test -bench . -benchtime=5s -count=3 命令，我们可以把测试进行3轮，结果如下:goos: darwin goarch: amd64 pkg: main cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkStrconv-12 217894554 31.76 ns/op BenchmarkStrconv-12 217140132 31.45 ns/op BenchmarkStrconv-12 219136828 31.79 ns/op BenchmarkFmtSprint-12 70683580 89.53 ns/op BenchmarkFmtSprint-12 63881758 82.51 ns/op BenchmarkFmtSprint-12 64984329 82.04 ns/op PASS ok main 54.296s结果变化也不大，看来strconv是真的比fmt.Sprint快很多。那快是快，会不会内存分配上情况就相反呢？通过go test -bench . -benchmem 这个命令我们可以看到两个方法的内存分配情况，结果如下：goos: darwin goarch: amd64 pkg: main cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkStrconv-12 43700922 27.46 ns/op 7 B/op 0 allocs/op BenchmarkFmtSprint-12 143412 80.88 ns/op 16 B/op 2 allocs/op PASS ok main 7.031s可以看到strconv在内存分配上是0次，每次运行使用的内存是7字节，只是fmt.Sprint的43.8%，简直是全方面的优于fmt.Sprint啊。那究竟是为什么strconv比fmt.Sprint好这么多呢？通过查看strconv的代码，我们发现，对于小于100的数字，strconv是直接通过digits和smallsString这两个常量进行转换的，而大于等于100的数字，则是通过不断除以100取余，然后再找到余数对应的字符串，把这些余数的结果拼起来进行转换的。const digits = "0123456789abcdefghijklmnopqrstuvwxyz" const smallsString = "00010203040506070809" + "10111213141516171819" + "20212223242526272829" + "30313233343536373839" + "40414243444546474849" + "50515253545556575859" + "60616263646566676869" + "70717273747576777879" + "80818283848586878889" + "90919293949596979899" // small returns the string for an i with 0 <= i < nSmalls. func small(i int) string { if i < 10 { return digits[i : i+1] } return smallsString[i*2 : i*2+2] } func formatBits(dst []byte, u uint64, base int, neg, append_ bool) (d []byte, s string) { ... for j := 4; j > 0; j-- { is := us % 100 * 2 us /= 100 i -= 2 a[i+1] = smallsString[is+1] a[i+0] = smallsString[is+0] } ... }而fmt.Sprint则是通过反射来实现这一目的的，fmt.Sprint得先判断入参的类型，在知道参数是int型后，再调用fmt.fmtInteger方法把int转换成string，这多出来的步骤肯定没有直接把int转成string来的高效。// fmtInteger formats signed and unsigned integers. func (f *fmt) fmtInteger(u uint64, base int, isSigned bool, verb rune, digits string) { ... switch base { case 10: for u >= 10 { i-- next := u / 10 buf[i] = byte('0' + u - next*10) u = next } ... }benchmark还有很多实用的函数，比如ResetTimer可以重置启动时耗费的准备时间，StopTimer和StartTimer则可以暂停和启动计时，让测试结果更集中在核心逻辑上。2.2 pprof2.2.1 使用介绍pprof是go语言官方提供的profile工具，支持可视化查看性能报告，功能十分强大。pprof基于定时器(10ms/次)对运行的go程序进行采样，搜集程序运行时的堆栈信息，包括CPU时间、内存分配等，最终生成性能报告。pprof有两个标准库，使用的场景不同：runtime/pprof 通过在代码中显式的增加触发和结束埋点来收集指定代码块运行时数据生成性能报告。net/http/pprof 是对runtime/pprof的二次封装，基于web服务运行，通过访问链接触发，采集服务运行时的数据生成性能报告。runtime/pprof的使用方法如下：package main import ( "os" "runtime/pprof" "time" ) func main() { w, _ := os.OpenFile("test_cpu", os.O_RDWR | os.O_CREATE | os.O_APPEND, 0644) pprof.StartCPUProfile(w) time.Sleep(time.Second) pprof.StopCPUProfile() }我们也可以使用另外一种方法，net/http/pprof：package main import ( "net/http" _ "net/http/pprof" ) func main() { err := http.ListenAndServe(":6060", nil) if err != nil { panic(err) } }将程序run起来后，我们通过访问http://127.0.0.1:6060/debug/pprof/就可以看到如下页面：点击profile就可以下载cpu profile文件。那我们如何查看我们的性能报告呢？ pprof支持两种查看模式，终端和web界面，注意: 想要查看可视化界面需要提前安装graphviz。这里我们以web界面为例，在终端内我们输入如下命令：go tool pprof -http :6060 test_cpu就会在浏览器里打开一个页面，内容如下：从界面左上方VIEW栏下，我们可以看到，pprof支持Flame Graph，dot Graph和Top等多种视图，下面我们将一一介绍如何阅读这些视图。2.2.1 火焰图 Flame Graph如何阅读首先，推荐直接阅读火焰图，在查函数耗时场景，这个比较直观；最简单的：横条越长，资源消耗、占用越多； 注意：每一个function 的横条虽然很长，但可能是他的下层“子调用”耗时产生的，所以一定要关注“下一层子调用”各自的耗时分布；每个横条支持点击下钻能力，可以更详细的分析子层的耗时占比。2.2.2 dot Graph 图如何阅读英文原文在这里：https://github.com/google/pprof/blob/master/doc/README.md节点颜色:红色表示耗时多的节点；绿色表示耗时少的节点；灰色表示耗时几乎可以忽略不计（接近零）；节点字体大小 :字体越大，表示占“上层函数调用”比例越大；（其实上层函数自身也有耗时，没包含在此）字体越小，表示占“上层函数调用”比例越小；线条（边）粗细:线条越粗，表示消耗了更多的资源；反之，则越少；线条（边）颜色:颜色越红，表示性能消耗占比越高；颜色越绿，表示性能消耗占比越低；灰色，表示性能消耗几乎可以忽略不计；虚线：表示中间有一些节点被“移除”或者忽略了；(一般是因为耗时较少所以忽略了) 实线：表示节点之间直接调用内联边标记：被调用函数已经被内联到调用函数中（对于一些代码行比较少的函数，编译器倾向于将它们在编译期展开从而消除函数调用，这种行为就是内联。）2.2.3 TOP 表如何阅读flat：当前函数，运行耗时（不包含内部调用其他函数的耗时）flat%：当前函数，占用的 CPU 运行耗时总比例（不包含外部调用函数）sum%：当前行的flat%与上面所有行的flat%总和。cum：当前函数加上它内部的调用的运行总耗时（包含内部调用其他函数的耗时）cum%：同上的 CPU 运行耗时总比例2.3 tracepprof已经有了对内存和CPU的分析能力，那trace工具有什么不同呢？虽然pprof的CPU分析器，可以告诉你什么函数占用了最多的CPU时间，但它并不能帮助你定位到是什么阻止了goroutine运行，或者在可用的OS线程上如何调度goroutines。这正是trace真正起作用的地方。我们需要更多关于Go应用中各个goroutine的执行情况的更为详细的信息，可以从P（goroutine调度器概念中的processor)和G（goroutine调度器概念中的goroutine）的视角完整的看到每个P和每个G在Tracer开启期间的全部“所作所为”，对Tracer输出数据中的每个P和G的行为分析并结合详细的event数据来辅助问题诊断的。Tracer可以帮助我们记录的详细事件包含有：与goroutine调度有关的事件信息：goroutine的创建、启动和结束；goroutine在同步原语（包括mutex、channel收发操作）上的阻塞与解锁。与网络有关的事件：goroutine在网络I/O上的阻塞和解锁；与系统调用有关的事件：goroutine进入系统调用与从系统调用返回；与垃圾回收器有关的事件：GC的开始/停止，并发标记、清扫的开始/停止。Tracer主要也是用于辅助诊断这三个场景下的具体问题的：并行执行程度不足的问题：比如没有充分利用多核资源等；因GC导致的延迟较大的问题；Goroutine执行情况分析，尝试发现goroutine因各种阻塞（锁竞争、系统调用、调度、辅助GC）而导致的有效运行时间较短或延迟的问题。2.3.1 trace性能报告打开trace性能报告，首页信息包含了多维度数据，如下图：View trace：以图形页面的形式渲染和展示tracer的数据，这也是我们最为关注/最常用的功能Goroutine analysis：以表的形式记录执行同一个函数的多个goroutine的各项trace数据Network blocking profile：用pprof profile形式的调用关系图展示网络I/O阻塞的情况Synchronization blocking profile：用pprof profile形式的调用关系图展示同步阻塞耗时情况Syscall blocking profile：用pprof profile形式的调用关系图展示系统调用阻塞耗时情况Scheduler latency profile：用pprof profile形式的调用关系图展示调度器延迟情况User-defined tasks和User-defined regions：用户自定义trace的task和regionMinimum mutator utilization：分析GC对应用延迟和吞吐影响情况的曲线图通常我们最为关注的是View trace和Goroutine analysis，下面将详细说说这两项的用法。2.3.2 view trace如果Tracer跟踪时间较长,trace会将View trace按时间段进行划分，避免触碰到trace-viewer的限制：View trace使用快捷键来缩放时间线标尺：w键用于放大（从秒向纳秒缩放），s键用于缩小标尺（从纳秒向秒缩放）。我们同样可以通过快捷键在时间线上左右移动：s键用于左移，d键用于右移。(游戏快捷键WASD)采样状态这个区内展示了三个指标：Goroutines、Heap和Threads，某个时间点上的这三个指标的数据是这个时间点上的状态快照采样：Goroutines：某一时间点上应用中启动的goroutine的数量，当我们点击某个时间点上的goroutines采样状态区域时（我们可以用快捷键m来准确标记出那个时间点），事件详情区会显示当前的goroutines指标采样状态：Heap指标则显示了某个时间点上Go应用heap分配情况（包括已经分配的Allocated和下一次GC的目标值NextGC）：Threads指标显示了某个时间点上Go应用启动的线程数量情况，事件详情区将显示处于InSyscall（整阻塞在系统调用上）和Running两个状态的线程数量情况：P视角区这里将View trace视图中最大的一块区域称为“P视角区”。这是因为在这个区域，我们能看到Go应用中每个P（Goroutine调度概念中的P）上发生的所有事件，包括：EventProcStart、EventProcStop、EventGoStart、EventGoStop、EventGoPreempt、Goroutine辅助GC的各种事件以及Goroutine的GC阻塞(STW)、系统调用阻塞、网络阻塞以及同步原语阻塞(mutex)等事件。除了每个P上发生的事件，我们还可以看到以单独行显示的GC过程中的所有事件。事件详情区点选某个事件后，关于该事件的详细信息便会在这个区域显示出来，事件详情区可以看到关于该事件的详细信息：Title：事件的可读名称；Start：事件的开始时间，相对于时间线上的起始时间；Wall Duration：这个事件的持续时间，这里表示的是G1在P4上此次持续执行的时间；Start Stack Trace：当P4开始执行G1时G1的调用栈；End Stack Trace：当P4结束执行G1时G1的调用栈；从上面End Stack Trace栈顶的函数为runtime.asyncPreempt来看，该Goroutine G1是被强行抢占了，这样P4才结束了其运行；Incoming flow：触发P4执行G1的事件；Outgoing flow：触发G1结束在P4上执行的事件；Preceding events：与G1这个goroutine相关的之前的所有的事件；Follwing events：与G1这个goroutine相关的之后的所有的事件All connected：与G1这个goroutine相关的所有事件。2.3.3 Goroutine analysisGoroutine analysis提供了从G视角看Go应用执行的图景。与View trace不同，这次页面中最广阔的区域提供的G视角视图，而不再是P视角视图。在这个视图中，每个G都会对应一个单独的条带（和P视角视图一样，每个条带都有两行），通过这一条带可以按时间线看到这个G的全部执行情况。通常仅需在goroutine analysis的表格页面找出执行最快和最慢的两个goroutine，在Go视角视图中沿着时间线对它们进行对比，以试图找出执行慢的goroutine究竟出了什么问题。2.4 后记虽然pprof和trace有着非常强大的profile能力，但在使用过程中，仍存在以下痛点：获取性能报告麻烦：一般大家做压测，为了更接近真实环境性能态，都使用生产环境/pre环境进行。而出于安全考虑，生产环境内网一般和PC办公内网是隔离不通的，需要单独配置通路才可以获得生产环境内网的profile 文件下载到PC办公电脑中，这也有一些额外的成本；查看profile分析报告麻烦：之前大家在本地查看profile 分析报告，一般 go tool pprof -http=":8083" profile 命令在本地PC开启一个web service 查看，并且需要至少安装graphviz 等库。查看trace分析同样麻烦：查看go trace 的profile 信息来分析routine 锁和生命周期时，也需要类似的方式在本地PC执行命令 go tool trace mytrace.profile 分享麻烦：如果我想把自己压测的性能结果内容，分享个另一位同学，那只能把1中获取的性能报告“profile文件”通过钉钉发给被分享人。然而有时候本地profile文件比较多，一不小心就发错了，还不如截图，但是截图又没有了交互放大、缩小、下钻等能力。处处不给力！留存复盘麻烦：系统的性能分析就像一份病历，每每看到阶段性的压测报告，总结或者对照时，不禁要询问，做过了哪些优化和改造，病因病灶是什么，有没有共性，值不值得总结归纳，现在是不是又面临相似的性能问题？那么能不能开发一个平台工具，解决以上的这些痛点呢？目前在阿里集团内部，高德的研发同学已经通过对go官方库的定制开发，实现了go语言性能平台，解决了以上这些痛点，并在内部进行了开源。该平台已面向阿里集团，累计实现性能场景快照数万条的获取和分析，解决了很多的线上服务性能调试和优化问题，这里暂时不展开，后续有机会可以单独分享。3. 性能调优-技巧篇除了前面提到的尽量用strconv而不是fmt.Sprint进行数字到字符串的转化以外，我们还将介绍一些在实际开发中经常会用到的技巧，供各位参考。3.1 字符串拼接拼接字符串为了书写方便快捷，最常用的两个方法是运算符 + 和 fmt.Sprintf()运算符 + 只能简单地完成字符串之间的拼接，fmt.Sprintf() 其底层实现使用了反射，性能上会有所损耗。从性能出发，兼顾易用可读，如果待拼接的变量不涉及类型转换且数量较少（<=5），拼接字符串推荐使用运算符 +，反之使用 fmt.Sprintf()。// 推荐：用+进行字符串拼接 func BenchmarkPlus(b *testing.B) { for i := 0; i < b.N; i++ { s := "a" + "b" _ = s } } // 不推荐：用fmt.Sprintf进行字符串拼接 func BenchmarkFmt(b *testing.B) { for i := 0; i < b.N; i++ { s := fmt.Sprintf("%s%s", "a", "b") _ = s } } goos: darwin goarch: amd64 pkg: main cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkPlus-12 1000000000 0.2658 ns/op 0 B/op 0 allocs/op BenchmarkFmt-12 16559949 70.83 ns/op 2 B/op 1 allocs/op PASS ok main 5.908s3.2 提前指定容器容量在初始化slice时，尽量指定容量，这是因为当添加元素时，如果容量的不足，slice会重新申请一个更大容量的容器，然后把原来的元素复制到新的容器中。// 推荐：初始化时指定容量 func BenchmarkGenerateWithCap(b *testing.B) { nums := make([]int, 0, 10000) for n := 0; n < b.N; n++ { for i:=0; i < 10000; i++ { nums = append(nums, i) } } } // 不推荐：初始化时不指定容量 func BenchmarkGenerate(b *testing.B) { nums := make([]int, 0) for n := 0; n < b.N; n++ { for i:=0; i < 10000; i++ { nums = append(nums, i) } } } goos: darwin goarch: amd64 pkg: main cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkGenerateWithCap-12 23508 336485 ns/op 476667 B/op 0 allocs/op BenchmarkGenerate-12 22620 68747 ns/op 426141 B/op 0 allocs/op PASS ok main 16.628s3.3 遍历 []struct{} 使用下标而不是 range常用的遍历方式有两种，一种是for循环下标遍历，一种是for循环range遍历，这两种遍历在性能上是否有差异呢？让我们来一探究竟。针对[]int，我们来看看两种遍历有和差别吧func getIntSlice() []int { nums := make([]int, 1024, 1024) for i := 0; i < 1024; i++ { nums[i] = i } return nums } // 用下标遍历[]int func BenchmarkIndexIntSlice(b *testing.B) { nums := getIntSlice() b.ResetTimer() for i := 0; i < b.N; i++ { var tmp int for k := 0; k < len(nums); k++ { tmp = nums[k] } _ = tmp } } // 用range遍历[]int元素 func BenchmarkRangeIntSlice(b *testing.B) { nums := getIntSlice() b.ResetTimer() for i := 0; i < b.N; i++ { var tmp int for _, num := range nums { tmp = num } _ = tmp } } goos: darwin goarch: amd64 pkg: demo/test cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkIndexIntSlice-12 3923230 270.2 ns/op 0 B/op 0 allocs/op BenchmarkRangeIntSlice-12 4518495 287.8 ns/op 0 B/op 0 allocs/op PASS ok demo/test 3.303s可以看到，在遍历[]int时，两种方式并无差别。我们再看看遍历[]struct{}的情况type Item struct { id int val [1024]byte } // 推荐：用下标遍历[]struct{} func BenchmarkIndexStructSlice(b *testing.B) { var items [1024]Item for i := 0; i < b.N; i++ { var tmp int for j := 0; j < len(items); j++ { tmp = items[j].id } _ = tmp } } // 推荐：用range的下标遍历[]struct{} func BenchmarkRangeIndexStructSlice(b *testing.B) { var items [1024]Item for i := 0; i < b.N; i++ { var tmp int for k := range items { tmp = items[k].id } _ = tmp } } // 不推荐：用range遍历[]struct{}的元素 func BenchmarkRangeStructSlice(b *testing.B) { var items [1024]Item for i := 0; i < b.N; i++ { var tmp int for _, item := range items { tmp = item.id } _ = tmp } } goos: darwin goarch: amd64 pkg: demo/test cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkIndexStructSlice-12 4413182 266.7 ns/op 0 B/op 0 allocs/op BenchmarkRangeIndexStructSlice-12 4545476 269.4 ns/op 0 B/op 0 allocs/op BenchmarkRangeStructSlice-12 33300 35444 ns/op 0 B/op 0 allocs/op PASS ok demo/test 5.282s可以看到，用for循环下标的方式性能都差不多，但是用range遍历数组里的元素时，性能则相差很多，前面两种方法是第三种方法的130多倍。主要原因是通过for k, v := range获取到的元素v实际上是原始值的一个拷贝。所以在面对复杂的struct进行遍历的时候，推荐使用下标。但是当遍历对象是复杂结构体的指针([]*struct{})时，用下标还是用range迭代元素的性能就差不多了。3.4 利用unsafe包避开内存copyunsafe包提供了任何类型的指针和 unsafe.Pointer 的相互转换及uintptr 类型和 unsafe.Pointer 可以相互转换，如下图unsafe包指针转换关系依据上述转换关系，其实除了string和[]byte的转换，也可以用于slice、map等的求长度及一些结构体的偏移量获取等，但是这种黑科技在一些情况下会带来一些匪夷所思的诡异问题，官方也不建议用，所以还是慎用，除非你确实很理解各种机制了，这里给出项目中实际用到的常规string和[]byte之间的转换，如下：func Str2bytes(s string) []byte { x := (*[2]uintptr)(unsafe.Pointer(&s)) h := [3]uintptr{x[0], x[1], x[1]} return *(*[]byte)(unsafe.Pointer(&h)) } func Bytes2str(b []byte) string { return *(*string)(unsafe.Pointer(&b)) } 我们通过benchmark来验证一下是否性能更优：// 推荐：用unsafe.Pointer实现string到bytes func BenchmarkStr2bytes(b *testing.B) { s := "testString" var bs []byte for n := 0; n < b.N; n++ { bs = Str2bytes(s) } _ = bs } // 不推荐：用类型转换实现string到bytes func BenchmarkStr2bytes2(b *testing.B) { s := "testString" var bs []byte for n := 0; n < b.N; n++ { bs = []byte(s) } _ = bs } // 推荐：用unsafe.Pointer实现bytes到string func BenchmarkBytes2str(b *testing.B) { bs := Str2bytes("testString") var s string b.ResetTimer() for n := 0; n < b.N; n++ { s = Bytes2str(bs) } _ = s } // 不推荐：用类型转换实现bytes到string func BenchmarkBytes2str2(b *testing.B) { bs := Str2bytes("testString") var s string b.ResetTimer() for n := 0; n < b.N; n++ { s = string(bs) } _ = s } goos: darwin goarch: amd64 pkg: demo/test cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkStr2bytes-12 1000000000 0.2938 ns/op 0 B/op 0 allocs/op BenchmarkStr2bytes2-12 38193139 28.39 ns/op 16 B/op 1 allocs/op BenchmarkBytes2str-12 1000000000 0.2552 ns/op 0 B/op 0 allocs/op BenchmarkBytes2str2-12 60836140 19.60 ns/op 16 B/op 1 allocs/op PASS ok demo/test 3.301s可以看到使用unsafe.Pointer比强制类型转换性能是要高不少的，从内存分配上也可以看到完全没有新的内存被分配。3.5 协程池go语言最大的特色就是很容易的创建协程，同时go语言的协程调度策略也让go程序可以最大化的利用cpu资源，减少线程切换。但是无限度的创建goroutine，仍然会带来问题。我们知道，一个go协程占用内存大小在2KB左右，无限度的创建协程除了会占用大量的内存空间，同时协程的切换也有不少开销，一次协程切换大概需要100ns，虽然相较于线程毫秒级的切换要优秀很多，但依然存在开销，而且这些协程最后还是需要GC来回收，过多的创建协程，对GC也是很大的压力。所以我们在使用协程时，可以通过协程池来限制goroutine数量，避免无限制的增长。限制协程的方式有很多，比如可以用channel来限制：var wg sync.WaitGroup ch := make(chan struct{}, 3) for i := 0; i < 10; i++ { ch <- struct{}{} wg.Add(1) go func(i int) { defer wg.Done() log.Println(i) time.Sleep(time.Second) <-ch }(i) } wg.Wait()这里通过限制channel长度为3，可以实现最多只有3个协程被创建的效果。当然也可以使用@烟渺实现的errgoup。使用方法如下：func Test_ErrGroupRun(t *testing.T) { errgroup := WithTimeout(nil, 10*time.Second) errgroup.SetMaxProcs(4) for index := 0; index < 10; index++ { errgroup.Run(nil, index, "test", func(context *gin.Context, i interface{}) (interface{}, error) { t.Logf("[%s]input:%+v, time:%s", "test", i, time.Now().Format("2006-01-02 15:04:05")) time.Sleep(2*time.Second) return i, nil }) } errgroup.Wait() }输出结果如下：=== RUN Test_ErrGroupRun errgroup_test.go:23: [test]input:0, time:2022-12-04 17:31:29 errgroup_test.go:23: [test]input:3, time:2022-12-04 17:31:29 errgroup_test.go:23: [test]input:1, time:2022-12-04 17:31:29 errgroup_test.go:23: [test]input:2, time:2022-12-04 17:31:29 errgroup_test.go:23: [test]input:4, time:2022-12-04 17:31:31 errgroup_test.go:23: [test]input:5, time:2022-12-04 17:31:31 errgroup_test.go:23: [test]input:6, time:2022-12-04 17:31:31 errgroup_test.go:23: [test]input:7, time:2022-12-04 17:31:31 errgroup_test.go:23: [test]input:8, time:2022-12-04 17:31:33 errgroup_test.go:23: [test]input:9, time:2022-12-04 17:31:33 --- PASS: Test_ErrGroupRun (6.00s) PASSerrgroup可以通过SetMaxProcs设定协程池的大小，从上面的结果可以看到，最多就4个协程在运行。3.6 sync.Pool 对象复用我们在代码中经常会用到json进行序列化和反序列化，举一个投放活动的例子，一个投放活动会有许多字段会转换为字节数组。type ActTask struct { Id int64 `ddb:"id"` // 主键id Status common.Status `ddb:"status"` // 状态 0=初始 1=生效 2=失效 3=过期 BizProd common.BizProd `ddb:"biz_prod"` // 业务类型 Name string `ddb:"name"` // 活动名 Adcode string `ddb:"adcode"` // 城市 RealTimeRuleByte []byte `ddb:"realtime_rule"` // 实时规则json ... } type RealTimeRuleStruct struct { Filter []*struct { PropertyId int64 `json:"property_id"` PropertyCode string `json:"property_code"` Operator string `json:"operator"` Value []string `json:"value"` } `json:"filter"` ExtData [1024]byte `json:"ext_data"` } func (at *ActTask) RealTimeRule() *form.RealTimeRule { if err := json.Unmarshal(at.RealTimeRuleByte, &at.RealTimeRuleStruct); err != nil { return nil } return at.RealTimeRuleStruct }以这里的实时投放规则为例，我们会将过滤规则反序列化为字节数组。每次json.Unmarshal都会申请一个临时的结构体对象，而这些对象都是分配在堆上的，会给 GC 造成很大压力，严重影响程序的性能。对于需要频繁创建并回收的对象，我们可以使用对象池来提升性能。sync.Pool可以将暂时不用的对象缓存起来，待下次需要的时候直接使用，不用再次经过内存分配，复用对象的内存，减轻 GC 的压力，提升系统的性能。sync.Pool的使用方法很简单，只需要实现 New 函数即可。对象池中没有对象时，将会调用 New 函数创建。var realTimeRulePool = sync.Pool{ New: func() interface{} { return new(RealTimeRuleStruct) }, }然后调用 Pool 的 Get() 和 Put() 方法来获取和放回池子中。rule := realTimeRulePool.Get().(*RealTimeRuleStruct) json.Unmarshal(buf, rule) realTimeRulePool.Put(rule)Get() 用于从对象池中获取对象，因为返回值是 interface{}，因此需要类型转换。Put() 则是在对象使用完毕后，放回到对象池。接下来我们进行性能测试，看看性能如何var realTimeRule = []byte("{\\\"filter\\\":[{\\\"property_id\\\":2,\\\"property_code\\\":\\\"search_poiid_industry\\\",\\\"operator\\\":\\\"in\\\",\\\"value\\\":[\\\"yimei\\\"]},{\\\"property_id\\\":4,\\\"property_code\\\":\\\"request_page_id\\\",\\\"operator\\\":\\\"in\\\",\\\"value\\\":[\\\"all\\\"]}],\\\"white_list\\\":[{\\\"property_id\\\":1,\\\"property_code\\\":\\\"white_list_for_adiu\\\",\\\"operator\\\":\\\"in\\\",\\\"value\\\":[\\\"j838ef77bf227chcl89888f3fb0946\\\",\\\"lb89bea9af558589i55559764bc83e\\\"]}],\\\"ipc_user_tag\\\":[{\\\"property_id\\\":1,\\\"property_code\\\":\\\"ipc_crowd_tag\\\",\\\"operator\\\":\\\"in\\\",\\\"value\\\":[\\\"test_20227041152_mix_ipc_tag\\\"]}],\\\"relation_id\\\":0,\\\"is_copy\\\":true}") // 推荐：复用一个对象，不用每次都生成新的 func BenchmarkUnmarshalWithPool(b *testing.B) { for n := 0; n < b.N; n++ { task := realTimeRulePool.Get().(*RealTimeRuleStruct) json.Unmarshal(realTimeRule, task) realTimeRulePool.Put(task) } } // 不推荐：每次都会生成一个新的临时对象 func BenchmarkUnmarshal(b *testing.B) { for n := 0; n < b.N; n++ { task := &RealTimeRuleStruct{} json.Unmarshal(realTimeRule, task) } } goos: darwin goarch: amd64 pkg: demo/test cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkUnmarshalWithPool-12 3627546 319.4 ns/op 312 B/op 7 allocs/op BenchmarkUnmarshal-12 2342208 490.8 ns/op 1464 B/op 8 allocs/op PASS ok demo/test 3.525s可以看到，两种方法在时间消耗上差不太多，但是在内存分配上差距明显，使用sync.Pool后内存占用仅为不使用的1/5。3.7 避免系统调用系统调用是一个很耗时的操作，在各种语言中都是，go也不例外，在go的GPM模型中，异步系统调用G会和MP分离，同步系统调用GM会和P分离，不管何种形式除了状态切换及内核态中执行操作耗时外，调度器本身的调度也耗时。所以在可以避免系统调用的地方尽量去避免// 推荐：不使用系统调用 func BenchmarkNoSytemcall(b *testing.B) { b.RunParallel(func(pb *testing.PB) { for pb.Next() { if configs.PUBLIC_KEY != nil { } } }) } // 不推荐：使用系统调用 func BenchmarkSytemcall(b *testing.B) { b.RunParallel(func(pb *testing.PB) { for pb.Next() { if os.Getenv("PUBLIC_KEY") != "" { } } }) } goos: darwin goarch: amd64 pkg: demo/test cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz BenchmarkNoSytemcall-12 1000000000 0.1495 ns/op 0 B/op 0 allocs/op BenchmarkSytemcall-12 37224988 31.10 ns/op 0 B/op 0 allocs/op PASS ok demo/test 1.877s4. 性能调优-实战篇案例1: go协程创建数据库连接不释放导致内存暴涨应用背景感谢@路现提供的案例。遇到的问题及表象特征线上机器偶尔出现内存使用率超过百分之九十报警。分析思路及排查方向在报警触发时，通过直接拉取线上应用的profile文件，查看内存分配情况，我们看到内存分配主要产生在本地缓存的组件上。但是分析代码并没有发现存在内存泄露的情况，看着像是资源一直没有被释放，进一步分析goroutine的profile文件发现存在大量的goroutine未释放，表现在本地缓存击穿后回源数据库，对数据库的查询访问一直不释放。调优手段与效果最终通过排查，发现使用的数据库组件存在bug，在极端情况下会出现死锁的情况，导致数据库访问请求无法返回也无法释放。最终bug修复后升级数据库组件版本解决了问题。案例2: 优惠索引内存分配大，gc 耗时高应用背景感谢@梅东提供的案例。遇到的问题及表象特征接口tp99高，偶尔会有一些特别耗时的请求，导致用户的优惠信息展示不出来分析思路及排查方向通过直接在平台上抓包观察，我们发现使用的分配索引这个方法占用的堆内存特别高，通过 top 可以看到是排在第一位的我们分析代码，可以看到，获取城市索引的地方，每次都是重新申请了内存的，通过改动为返回指针，就不需要每次都单独申请内存了，核心代码改动：调优手段与效果修改后，上线观察，可以看到使用中的内存以及gc耗时都有了明显降低案例3：流量上涨导致cpu异常飙升应用背景感谢@君度提供的案例。遇到的问题及表象特征能量站v2接口和task-home-page接口流量较大时，会造成ab实验策略匹配时cpu飙升分析思路及排查方向调优手段与效果主要优化点如下：1、优化toEntity方法，简化为单独的ID()方法2、优化数组、map初始化结构3、优化adCode转换为string过程4、关闭过多的match log打印优化后profile：优化上线前后CPU的对比案例4： 内存对象未释放导致内存泄漏应用背景感谢@淳深提供的案例，提供案例的服务，日常流量峰值在百万qps左右，是高德内部十分重要的服务。此前该服务是由java实现的，后来用go语言进行重构，在重构完成切全量后，有许多性能优化的优秀案例，这里选取内存泄漏相关的一个案例分享给大家，希望对大家在自己服务进行内存泄漏问题排查时能提供参考和帮助。遇到的问题及表象特征go语言版本全量切流后，每天会对服务各项指标进行详细review，发现每日内存上涨约0.4%，如下图在go版本服务切全量前，从第一张图可以看到整个内存使用是平稳的，无上涨趋势，但是切go版本后，从第二张图可以看到，整个内存曲线呈上升趋势，遂认定内存泄漏，开始排查内存泄漏的“罪魁祸首”。分析思路及排查方向我们先到线上机器抓取当前时间的heap文件，间隔一天后再次抓取heap文件，通过pprof diff对比，我们发现time.NewTicker的内存占用增长了几十MB(由于未保留当时的heap文件，此处没有截图)，通过调用栈信息，我们找到了问题的源头，来自中间件vipserver client的SrvHost方法，通过深扒vipserver client代码，我们发现，每个vipserver域名都会有一个对应的协程，这个协程每隔三秒钟就会新建一个ticker对象，且用过的ticker对象没有stop，也就不会释放相应的内存资源。而这个time.NewTicker会创建一个timer对象，这个对象会占用72字节内存。在服务运行一天的情况下，进过计算，该对象累计会在内存中占用约35.6MB，和上述内存每日增长0.4%正好能对上，我们就能断定这个内存泄漏来自这里。调优手段与效果知道是timer对象重复创建的问题后，只需要修改这部分的代码就好了，最新的vipserver client修改了此处的逻辑，如下修改完后，运行一段时间，内存运行状态平稳，已无内存泄漏问题。结语目前go语言不仅在阿里集团内部，在整个互联网行业内也越来越流行，希望本文能为正在使用go语言的同学在性能优化方面带来一些参考价值。在阿里集团内部，高德也是最早规模化使用go语言的团队之一，目前高德线上运行的go服务已经达到近百个，整体qps已突破百万量级。在使用go语言的同时，高德也为集团内go语言生态建设做出了许多贡献，包括开发支持阿里集团常见的中间件（比如配置中心-Diamond、分布式RPC服务框架-HSF、服务发现-Vipserver、消息队列-MetaQ、流量控制-Sentinel、日志追踪-Eagleeye等）go语言版本，并被阿里中间件团队官方收录。但是go语言生态建设仍然有很长的道路要走，希望能有更多对go感兴趣的同学能够加入我们，一起参与阿里的go生态建设，乃至为互联网业界的go生态发展添砖加瓦。

文章

设计模式 · 缓存 · Java · 中间件 · 测试技术 · Go · 调度 · 数据库 · 索引 · 容器

2023-03-03

使用 DataAnnotations（数据注解）实现模型的通用数据校验

参数校验的意义在实际项目开发中，无论任何方式、任何规模的开发模式，项目中都离不开对接入数据模型参数的合法性校验，目前普片的开发模式基本是前后端分离，当用户在前端页面中输入一些表单数据时，点击提交按钮，触发请求目标服务器的一系列后续操作，在这中间的执行过程中（标准做法推荐）无论是前端代码部分，还是服务端代码部分都应该有针对用户输入数据的合法性校验，典型做法如下：前端部分：当用户在页面输入表单数据时，前端监听页面表单事件触发相应的数据合法性校验规则，当数据非法时，合理的提示用户数据错误，只有当所有表单数据都校验通过后，才继续提交数据给目标后端对应的接口；后端部分：当前端数据合法校验通过后，向目标服务器提交表单数据时，服务端接收到相应的提交数据，在入口源头出就应该触发相关的合法性校验规则，当数据都校验通过后，继续执行后续的相关业务逻辑处理，反之则响应相关非法数据的提示信息；特别说明：在实际的项目中，无论前端部分还是服务端部分，参数的校验都是很有必要性的。无效的参数，可能会导致应用程序的异常和一些不可预知的错误行为。常用参数的校验这里例举一些项目中比较常用的参数模型校验项，如下所示：Name：姓名校验，比如需要是纯汉字的姓名；Password：密码强度验证，比如要求用户输入必须包含大小写字母、数字和特殊符号的强密码；QQ号：QQ 号码验证，是否是有效合法的 QQ 号码；China Postal Code：中国邮政编码；IP Address：IPV4 或者 IPV6 地址验证；Phone：手机号码或者座机号码合法性验证；ID Card：身份证号码验证，比如：15 位和 18 位数身份证号码；Email Address：邮箱地址的合法性校验；String：字符串验证，比如字段是否不为 null、长度是否超限；URL：验证属性是否具有 URL 格式；Number：数值型参数校验，数值范围校验，比如非负数，非负整数，正整数等；File：文件路径及扩展名校验；对于参数校验，常见的方式有正则匹配校验，通过对目标参数编写合法的正则表达式，实现对参数合法性的校验。.NET 中内置 DataAnnotations 提供的特性校验上面我们介绍了一些常用的参数验证项，接下来我们来了解下在 .NET 中内置提供的 DataAnnotations 数据注解，该类提供了一些常用的验证参数特性。官方解释：提供用于为 ASP.NET MVC 和 ASP.NET 数据控件定义元数据的特性类。该类位于 System.ComponentModel.DataAnnotations 命名空间。关于 DataAnnotations 中的特性介绍让我们可以通过这些特性对 API 请求中的参数进行验证，常用的特性一般有：[ValidateNever]：指示应从验证中排除属性或参数。[CreditCard]：验证属性是否具有信用卡格式。[Compare]：验证模型中的两个属性是否匹配。[EmailAddress]：验证属性是否具有电子邮件格式。[Phone]：验证属性是否具有电话号码格式。[Range]：验证属性值是否位于指定范围内。[RegularExpression]：验证属性值是否与指定的正则表达式匹配。[Required]：验证字段是否不为 null。[StringLength]：验证字符串属性值是否不超过指定的长度限制。[Url]：验证属性是否具有 URL 格式。其中 RegularExpression 特性，基于正则表达式可以扩展实现很多常用的验证类型，下面的（基于 DataAnnotations 的通用模型校验封装）环节举例说明；关于该类更多详细信息请查看，https://learn.microsoft.com/zh-cn/dotnet/api/system.componentmodel.dataannotations?view=net-7.0基于 DataAnnotations 的通用模型校验封装此处主要是使用了 Validator.TryValidateObject() 方法：Validator.TryValidateObject(object instance, ValidationContext validationContext, ICollection<ValidationResult>? validationResults, bool validateAllProperties);Validator 类提供如下校验方法：基于 DataAnnotations 的特性校验助手实现步骤错误成员对象类 ErrorMembernamespace Jeff.Common.Validatetion; /// <summary> /// 错误成员对象 /// </summary> public class ErrorMember { /// <summary> /// 错误信息 /// </summary> public string? ErrorMessage { get; set; } /// <summary> /// 错误成员名称 /// </summary> public string? ErrorMemberName { get; set; } }验证结果类 ValidResultnamespace Jeff.Common.Validatetion; /// <summary> /// 验证结果类 /// </summary> public class ValidResult { public ValidResult() { ErrorMembers = new List<ErrorMember>(); } /// <summary> /// 错误成员列表 /// </summary> public List<ErrorMember> ErrorMembers { get; set; } /// <summary> /// 验证结果 /// </summary> public bool IsVaild { get; set; } }定义操作正则表达式的公共类 RegexHelper（基于 RegularExpression 特性扩展）using System; using System.Net; using System.Text.RegularExpressions; namespace Jeff.Common.Validatetion; /// <summary> /// 操作正则表达式的公共类 /// Regex 用法参考：https://learn.microsoft.com/zh-cn/dotnet/api/system.text.regularexpressions.regex.-ctor?redirectedfrom=MSDN&view=net-7.0 /// </summary> public class RegexHelper { #region 常用正则验证模式字符串 public enum ValidateType { Email, // 邮箱 TelePhoneNumber, // 固定电话（座机） MobilePhoneNumber, // 移动电话 Age, // 年龄（1-120 之间有效） Birthday, // 出生日期 Timespan, // 时间戳 IdentityCardNumber, // 身份证 IpV4, // IPv4 地址 IpV6, // IPV6 地址 Domain, // 域名 English, // 英文字母 Chinese, // 汉字 MacAddress, // MAC 地址 Url, // URL } private static readonly Dictionary<ValidateType, string> keyValuePairs = new Dictionary<ValidateType, string> { { ValidateType.Email, _Email }, { ValidateType.TelePhoneNumber,_TelephoneNumber }, { ValidateType.MobilePhoneNumber,_MobilePhoneNumber }, { ValidateType.Age,_Age }, { ValidateType.Birthday,_Birthday }, { ValidateType.Timespan,_Timespan }, { ValidateType.IdentityCardNumber,_IdentityCardNumber }, { ValidateType.IpV4,_IpV4 }, { ValidateType.IpV6,_IpV6 }, { ValidateType.Domain,_Domain }, { ValidateType.English,_English }, { ValidateType.Chinese,_Chinese }, { ValidateType.MacAddress,_MacAddress }, { ValidateType.Url,_Url }, }; public const string _Email = @"^(\w)+(\.\w)*@(\w)+((\.\w+)+)$"; // ^[\w-]+(\.[\w-]+)*@[\w-]+(\.[\w-]+)+$ , [A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,4} public const string _TelephoneNumber = @"(d+-)?(d{4}-?d{7}|d{3}-?d{8}|^d{7,8})(-d+)?"; //座机号码（中国大陆） public const string _MobilePhoneNumber = @"^(13[0-9]|14[01456879]|15[0-35-9]|16[2567]|17[0-8]|18[0-9]|19[0-35-9])\d{8}$"; //移动电话 public const string _Age = @"^(?:[1-9][0-9]?|1[01][0-9]|120)$"; // 年龄 1-120 之间有效 public const string _Birthday = @"^((?:19[2-9]\d{1})|(?:20(?:(?:0[0-9])|(?:1[0-8]))))((?:0?[1-9])|(?:1[0-2]))((?:0?[1-9])|(?:[1-2][0-9])|30|31)$"; public const string _Timespan = @"^15|16|17\d{8,11}$"; // 目前时间戳是15开头，以后16、17等开头，长度 10 位是秒级时间戳的正则，13 位时间戳是到毫秒级的。 public const string _IdentityCardNumber = @"^[1-9]\d{7}((0\d)|(1[0-2]))(([0|1|2]\d)|3[0-1])\d{3}$|^[1-9]\d{5}[1-9]\d{3}((0\d)|(1[0-2]))(([0|1|2]\d)|3[0-1])\d{3}([0-9]|X)$"; public const string _IpV4 = @"^((2(5[0-5]|[0-4]\d))|[0-1]?\d{1,2})(\.((2(5[0-5]|[0-4]\d))|[0-1]?\d{1,2})){3}$"; public const string _IpV6 = @"^\s*((([0-9A-Fa-f]{1,4}:){7}([0-9A-Fa-f]{1,4}|:))|(([0-9A-Fa-f]{1,4}:){6}(:[0-9A-Fa-f]{1,4}|((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){5}(((:[0-9A-Fa-f]{1,4}){1,2})|:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){4}(((:[0-9A-Fa-f]{1,4}){1,3})|((:[0-9A-Fa-f]{1,4})?:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){3}(((:[0-9A-Fa-f]{1,4}){1,4})|((:[0-9A-Fa-f]{1,4}){0,2}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){2}(((:[0-9A-Fa-f]{1,4}){1,5})|((:[0-9A-Fa-f]{1,4}){0,3}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){1}(((:[0-9A-Fa-f]{1,4}){1,6})|((:[0-9A-Fa-f]{1,4}){0,4}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(:(((:[0-9A-Fa-f]{1,4}){1,7})|((:[0-9A-Fa-f]{1,4}){0,5}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:)))(%.+)?\s*$"; public const string _Domain = @"^[a-zA-Z0-9][-a-zA-Z0-9]{0,62}(\.[a-zA-Z0-9][-a-zA-Z0-9]{0,62})+\.?$"; public const string _English = @"^[A-Za-z]+$"; public const string _Chinese = @"^[\u4e00-\u9fa5]{0,}$"; public const string _MacAddress = @"^([0-9A-F]{2})(-[0-9A-F]{2}){5}$"; public const string _Url = @"^[a-zA-z]+://(\w+(-\w+)*)(\.(\w+(-\w+)*))*(\?\S*)?$"; #endregion /// <summary> /// 获取验证模式字符串 /// </summary> /// <param name="validateType"></param> /// <returns></returns> public static (bool hasPattern, string pattern) GetValidatePattern(ValidateType validateType) { bool hasPattern = keyValuePairs.TryGetValue(validateType, out string? pattern); return (hasPattern, pattern ?? string.Empty); } #region 验证输入字符串是否与模式字符串匹配 /// <summary> /// 验证输入字符串是否与模式字符串匹配 /// </summary> /// <param name="input">输入的字符串</param> /// <param name="validateType">模式字符串类型</param> /// <param name="matchTimeout">超时间隔</param> /// <param name="options">筛选条件</param> /// <returns></returns> public static (bool isMatch, string info) IsMatch(string input, ValidateType validateType, TimeSpan matchTimeout, RegexOptions options = RegexOptions.None) { var (hasPattern, pattern) = GetValidatePattern(validateType); if (hasPattern && !string.IsNullOrWhiteSpace(pattern)) { bool isMatch = IsMatch(input, pattern, matchTimeout, options); if (isMatch) return (true, "Format validation passed."); // 格式验证通过。 else return (false, "Format validation failed."); // 格式验证未通过。 } return (false, "Unknown ValidatePattern."); // 未知验证模式 } /// <summary> /// 验证输入字符串是否与模式字符串匹配，匹配返回true /// </summary> /// <param name="input">输入字符串</param> /// <param name="pattern">模式字符串</param> /// <returns></returns> public static bool IsMatch(string input, string pattern) { return IsMatch(input, pattern, TimeSpan.Zero, RegexOptions.IgnoreCase); } /// <summary> /// 验证输入字符串是否与模式字符串匹配，匹配返回true /// </summary> /// <param name="input">输入的字符串</param> /// <param name="pattern">模式字符串</param> /// <param name="matchTimeout">超时间隔</param> /// <param name="options">筛选条件</param> /// <returns></returns> public static bool IsMatch(string input, string pattern, TimeSpan matchTimeout, RegexOptions options = RegexOptions.None) { return Regex.IsMatch(input, pattern, options, matchTimeout); } #endregion }定义验证结果统一模型格式类 ResponseInfo（此类通常也是通用的数据响应模型类）namespace Jeff.Common.Model; public sealed class ResponseInfo<T> where T : class { /* Microsoft.AspNetCore.Http.StatusCodes System.Net.HttpStatusCode */ /// <summary> /// 响应代码（自定义） /// </summary> public int Code { get; set; } /// <summary> /// 接口状态 /// </summary> public bool Success { get; set; } #region 此处可以考虑多语言国际化设计（语言提示代号对照表） /// <summary> /// 语言对照码，参考：https://blog.csdn.net/shenenhua/article/details/79150053 /// </summary> public string Lang { get; set; } = "zh-cn"; /// <summary> /// 提示信息 /// </summary> public string Message { get; set; } = string.Empty; #endregion /// <summary> /// 数据体 /// </summary> public T? Data { get; set; } }实现验证助手类 ValidatetionHelper，配合 System.ComponentModel.DataAnnotations 类使用// 数据注解，https://learn.microsoft.com/zh-cn/dotnet/api/system.componentmodel.dataannotations?view=net-7.0 using System.ComponentModel.DataAnnotations; using Jeff.Common.Model; namespace Jeff.Common.Validatetion; /// <summary> /// 验证助手类 /// </summary> public sealed class ValidatetionHelper { /// <summary> /// DTO 模型校验 /// </summary> /// <param name="value"></param> /// <returns></returns> public static ValidResult IsValid(object value) { var result = new ValidResult(); try { var validationContext = new ValidationContext(value); var results = new List<ValidationResult>(); bool isValid = Validator.TryValidateObject(value, validationContext, results, true); result.IsVaild = isValid; if (!isValid) { foreach (ValidationResult? item in results) { result.ErrorMembers.Add(new ErrorMember() { ErrorMessage = item.ErrorMessage, ErrorMemberName = item.MemberNames.FirstOrDefault() }); } } } catch (ValidationException ex) { result.IsVaild = false; result.ErrorMembers = new List<ErrorMember> { new ErrorMember() { ErrorMessage = ex.Message, ErrorMemberName = "Internal error" } }; } return result; } /// <summary> /// DTO 模型校验统一响应信息 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="model"></param> /// <returns></returns> public static ResponseInfo<ValidResult> GetValidInfo<T>(T model) where T : class { var result = new ResponseInfo<ValidResult>(); var validResult = IsValid(model); if (!validResult.IsVaild) { result.Code = 420; result.Message = "DTO 模型参数值异常"; result.Success = false; result.Data = validResult; } else { result.Code = 200; result.Success = true; result.Message = "DTO 模型参数值合法"; } return result; } }如何使用 DataAnnotations 封装的特性校验助手？首先定义一个数据模型类（DTO），添加校验特性 ValidationAttributeusing System.ComponentModel.DataAnnotations; using Jeff.Common.Validatetion; namespace Jeff.Comm.Test; public class Person { [Display(Name = "姓名"), Required(ErrorMessage = "{0}必须填写")] public string Name { get; set; } [Display(Name = "邮箱")] [Required(ErrorMessage = "{0}必须填写")] [RegularExpression(RegexHelper._Email, ErrorMessage = "RegularExpression: {0}格式非法")] [EmailAddress(ErrorMessage = "EmailAddress: {0}格式非法")] public string Email { get; set; } [Display(Name = "Age年龄")] [Required(ErrorMessage = "{0}必须填写")] [Range(1, 120, ErrorMessage = "超出范围")] [RegularExpression(RegexHelper._Age, ErrorMessage = "{0}超出合理范围")] public int Age { get; set; } [Display(Name = "Birthday出生日期")] [Required(ErrorMessage = "{0}必须填写")] [RegularExpression(RegexHelper._Timespan, ErrorMessage = "{0}超出合理范围")] public TimeSpan Birthday { get; set; } [Display(Name = "Address住址")] [Required(ErrorMessage = "{0}必须填写")] [StringLength(200, MinimumLength = 10, ErrorMessage = "{0}输入长度不正确")] public string Address { get; set; } [Display(Name = "Mobile手机号码")] [Required(ErrorMessage = "{0}必须填写")] [RegularExpression(RegexHelper._MobilePhoneNumber, ErrorMessage = "{0}格式非法")] public string Mobile { get; set; } [Display(Name = "Salary薪水")] [Required(ErrorMessage = "{0}必须填写")] [Range(typeof(decimal), "1000.00", "3000.99")] public decimal Salary { get; set; } [Display(Name = "MyUrl连接")] [Required(ErrorMessage = "{0}必须填写")] [Url(ErrorMessage = "Url：{0}格式非法")] [RegularExpression(RegexHelper._Url, ErrorMessage = "RegularExpression：{0}格式非法")] public string MyUrl { get; set; } }控制台调用通用校验助手验证方法 ValidatetionHelper.IsValid() 或 ValidatetionHelper.GetValidInfo()// 通用模型数据验证测试 static void ValidatetionTest() { var p = new Person { Name = "", Age = -10, Email = "www.baidu.com", MobilePhoneNumber = "12345", Salary = 4000, MyUrl = "aaa" }; // 调用通用模型校验 var result = ValidatetionHelper.IsValid(p); if (!result.IsVaild) { foreach (ErrorMember errorMember in result.ErrorMembers) { // 控制台打印字段验证信息 Console.WriteLine($"{errorMember.ErrorMemberName}：{errorMember.ErrorMessage}"); } } Console.WriteLine(); // 调用通用模型校验，返回统一数据格式 var validInfo = ValidatetionHelper.GetValidInfo(p); var options = new JsonSerializerOptions { Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping, // 设置中文编码乱码 WriteIndented = false }; string jsonStr = JsonSerializer.Serialize(validInfo, options); Console.WriteLine($"校验结果返回统一数据格式：{jsonStr}"); }在控制台Program.Main 方法中调用 ValidatetionTest() 方法：internal class Program { static void Main(string[] args) { Console.WriteLine("Hello, World!"); { #region 数据注解（DataAnnotations）模型验证 ValidatetionTest(); #endregion } Console.ReadKey(); }启动控制台，输出如下信息：如何实现自定义的验证特性？当我们碰到这些参数需要验证的时候，而上面内置类提供的特性不能满足需求时，此时我们可以实现自定义的验证特性来满足校验需求，按照微软给出的编码规则，我们只需继承 ValidationAttribute 类，并重写 IsValid() 方法即可。自定义校验特性案例比如实现一个密码强度的验证，实现步骤如下：定义密码强度规则，只包含英文字母、数字和特殊字符的组合，并且组合长度至少 8 位数；/// <summary> /// 只包含英文字母、数字和特殊字符的组合 /// </summary> /// <returns></returns> public static bool IsCombinationOfEnglishNumberSymbol(string input, int? minLength = null, int? maxLength = null) { var pattern = @"(?=.*\d)(?=.*[a-zA-Z])(?=.*[^a-zA-Z\d])."; if (minLength is null && maxLength is null) pattern = $@"^{pattern}+$"; else if (minLength is not null && maxLength is null) pattern = $@"^{pattern}{{{minLength},}}$"; else if (minLength is null && maxLength is not null) pattern = $@"^{pattern}{{1,{maxLength}}}$"; else pattern = $@"^{pattern}{{{minLength},{maxLength}}}$"; return Regex.IsMatch(input, pattern); }实现自定义特性 EnglishNumberSymbolCombinationAttribute，继承自 ValidationAttribute；using System.ComponentModel.DataAnnotations; namespace Jeff.Common.Validatetion.CustomAttributes; /// <summary> /// 是否是英文字母、数字和特殊字符的组合 /// </summary> public class EnglishNumberSymbolCombinationAttribute : ValidationAttribute { /// <summary> /// 默认的错误提示信息 /// </summary> private const string error = "无效的英文字母、数字和特殊字符的组合"; protected override ValidationResult IsValid(object value, ValidationContext validationContext) { if (value is null) return new ValidationResult("参数值为 null"); //if (value is null) //{ // throw new ArgumentNullException(nameof(attribute)); //} // 验证参数逻辑 value 是需要验证的值，而 validationContext 中包含了验证相关的上下文信息，这里可自己封装一个验证格式的 FormatValidation 类 if (FormatValidation.IsCombinationOfEnglishNumberSymbol(value as string, 8)) //验证成功返回 success return ValidationResult.Success; //不成功提示验证错误的信息 else return new ValidationResult(ErrorMessage ?? error); } }以上就实现了一个自定义规则的自定义验证特性，使用方式很简单，可以把它附属在我们请求的参数上或者 DTO 里的属性，也可以是 Action 上的形参，如下所示：public class CreateDTO { [Required] public string StoreName { get; init; } [Required] // 附属在 DTO 里的属性 [EnglishNumberSymbolCombination(ErrorMessage = "UserId 必须是英文字母、数字和特殊符号的组合")] public string UserId { get; init; } } ... // 附属在 Action 上的形参 [HttpGet] public async ValueTask<ActionResult> Delete([EnglishNumberSymbolCombination]string userId, string storeName)该自定义验证特性还可以结合 DataAnnotations 内置的 [Compare] 特性，可以实现账号注册的密码确认验证（输入密码和确认密码是否一致性）。关于更多自定义参数校验特性，感兴趣的小伙伴可参照上面案例的实现思路，自行扩展实现哟。总结对于模型参数的校验，在实际项目系统中是非常有必要性的（通常在数据源头提供验证），利用 .NET 内置的 DataAnnotations（数据注解）提供的特性校验，可以很方便的实现通用的模型校验助手，关于其他特性的用法，请自行参考微软官方文档，这里注意下RegularExpressionAttribute（指定 ASP.NET 动态数据中的数据字段值必须与指定的正则表达式匹配），该特性可以方便的接入正则匹配验证，当遇到复杂的参数校验时，可以快速方便的扩展自定义校验特性，从此告别传统编码中各种 if(xxx != yyyy) 判断的验证，让整体代码编写更佳简练干净。

文章

开发框架 · 前端开发 · 网络协议 · .NET · API · 数据安全/隐私保护

2023-02-27

应用场景系列之（1）：流量管理下的熔断场景

在采用Istio 服务网格技术的很多客户案例中, 熔断是其中一个非常普遍的流量管理场景。在启用网格技术之前, 在一些使用Resilience4j 的Java 服务中客户已经使用了熔断功能, 但相比之下, Istio能够在网络级别支持启用熔断能力，无需集成到每个服务的应用程序代码中。深入理解熔断器在不同场景下的行为, 是将其应用到线上环境之前的关键前提。介绍启用熔断功能，需要创建一个目标规则来为目标服务配置熔断。其中, connectionPool下定义了与熔断功能相关的参数, 相关配置参数为：tcp.maxConnections: 到目标主机的最大 HTTP1 /TCP 连接数。默认值为2³²-1。http.http1MaxPendingRequests：等待就绪的连接池连接时，最多可以排队的请求数量。默认值为1024。http.http2MaxRequests：对后端服务的最大活跃请求数。默认值为1024。这些参数在一个简单的场景中, 如一个客户端和一个目标服务实例（在 Kubernetes 环境中，一个实例相当于一个 pod）的情况下是清晰的。然而, 在生产环境中，比较可能出现的场景是：一个客户端实例和多个目标服务实例多个客户端实例和单个目标服务实例客户端和目标服务的多个实例我们创建了两个 python 脚本——一个用于表示目标服务，另一个用于调用服务的客户端。服务器脚本是一个简单的 Flask 应用程序，它公开一个休眠 5 秒的API服务端点，然后返回一个“hello world!”字符串。示例代码如下所示：#! /usr/bin/env python3 from flask import Flask import time app = Flask(__name__) @app.route('/hello') def get(): time.sleep(5) return 'hello world!' if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port='9080', threaded = True)客户端脚本以 10 个为一组调用服务器端点，即 10 个并行请求，然后在发送下一批 10 个请求之前休眠一段时间。它在无限循环中执行此操作。为了确保当我们运行客户端的多个pod 时，它们都同时发送批处理，我们使用系统时间（每分钟的第 0、20 和40秒）发送批处理。#! /usr/bin/env python3 import requests import time import sys from datetime import datetime import _thread def timedisplay(t): return t.strftime("%H:%M:%S") def get(url): try: stime = datetime.now() start = time.time() response = requests.get(url) etime = datetime.now() end = time.time() elapsed = end-start sys.stderr.write("Status: " + str(response.status_code) + ", Start: " + timedisplay(stime) + ", End: " + timedisplay(etime) + ", Elapsed Time: " + str(elapsed)+"\n") sys.stdout.flush() except Exception as myexception: sys.stderr.write("Exception: " + str(myexception)+"\n") sys.stdout.flush() time.sleep(30) while True: sc = int(datetime.now().strftime('%S')) time_range = [0, 20, 40] if sc not in time_range: time.sleep(1) continue sys.stderr.write("\n----------Info----------\n") sys.stdout.flush() # Send 10 requests in parallel for i in range(10): _thread.start_new_thread(get, ("http://circuit-breaker-sample-server:9080/hello", )) time.sleep(2)部署示例应用使用如下YAML部署示例应用,################################################################################################## # circuit-breaker-sample-server services ################################################################################################## apiVersion: v1 kind: Service metadata: name: circuit-breaker-sample-server labels: app: circuit-breaker-sample-server service: circuit-breaker-sample-server spec: ports: - port: 9080 name: http selector: app: circuit-breaker-sample-server --- apiVersion: apps/v1 kind: Deployment metadata: name: circuit-breaker-sample-server labels: app: circuit-breaker-sample-server version: v1 spec: replicas: 1 selector: matchLabels: app: circuit-breaker-sample-server version: v1 template: metadata: labels: app: circuit-breaker-sample-server version: v1 spec: containers: - name: circuit-breaker-sample-server image: registry.cn-hangzhou.aliyuncs.com/acs/istio-samples:circuit-breaker-sample-server.v1 imagePullPolicy: Always ports: - containerPort: 9080 --- ################################################################################################## # circuit-breaker-sample-client services ################################################################################################## apiVersion: apps/v1 kind: Deployment metadata: name: circuit-breaker-sample-client labels: app: circuit-breaker-sample-client version: v1 spec: replicas: 1 selector: matchLabels: app: circuit-breaker-sample-client version: v1 template: metadata: labels: app: circuit-breaker-sample-client version: v1 spec: containers: - name: circuit-breaker-sample-client image: registry.cn-hangzhou.aliyuncs.com/acs/istio-samples:circuit-breaker-sample-client.v1 imagePullPolicy: Always 启动之后, 可以看到为客户端和服务器端分别启动了对应的pod, 类似如下：> kubectl get po |grep circuit circuit-breaker-sample-client-d4f64d66d-fwrh4 2/2 Running 0 1m22s circuit-breaker-sample-server-6d6ddb4b-gcthv 2/2 Running 0 1m22s在未定义目标规则限制的情况下, 服务器端可以满足处理并发的10个客户端请求, 因此在服务器端的响应结果始终是200。客户端侧的日志应当类似如下：----------Info---------- Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.016539812088013 Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.012614488601685 Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.015984535217285 Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.015599012374878 Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.012874364852905 Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.018714904785156 Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.010422468185425 Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.012431621551514 Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.011001348495483 Status: 200, Start: 02:39:20, End: 02:39:25, Elapsed Time: 5.01432466506958启用熔断规则通过服务网格技术启用熔断规则, 只需要针对目标服务定义对应的目标规则DestinationRule即可。创建并应用以下目标规则：apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: circuit-breaker-sample-server spec: host: circuit-breaker-sample-server trafficPolicy: connectionPool: tcp: maxConnections: 5它将与目标服务的 TCP 连接数限制为 5。让我们看看它在不同场景中的工作方式。场景#1：一个客户端实例和一个目标服务实例在这种情况下，客户端和目标服务都只有一个 pod。当我们启动客户端 pod 并监控日志时（建议重启下客户端以获得更直观的统计结果），我们会看到类似以下内容：----------Info---------- Status: 200, Start: 02:49:40, End: 02:49:45, Elapsed Time: 5.0167787075042725 Status: 200, Start: 02:49:40, End: 02:49:45, Elapsed Time: 5.011920690536499 Status: 200, Start: 02:49:40, End: 02:49:45, Elapsed Time: 5.017078161239624 Status: 200, Start: 02:49:40, End: 02:49:45, Elapsed Time: 5.018405437469482 Status: 200, Start: 02:49:40, End: 02:49:45, Elapsed Time: 5.018689393997192 Status: 200, Start: 02:49:40, End: 02:49:50, Elapsed Time: 10.018936395645142 Status: 200, Start: 02:49:40, End: 02:49:50, Elapsed Time: 10.016417503356934 Status: 200, Start: 02:49:40, End: 02:49:50, Elapsed Time: 10.019930601119995 Status: 200, Start: 02:49:40, End: 02:49:50, Elapsed Time: 10.022735834121704 Status: 200, Start: 02:49:40, End: 02:49:55, Elapsed Time: 15.02303147315979可以看到所有请求都成功。但是，每批中只有 5 个请求的响应时间约为 5 秒，其余的要慢得多（大部分为10秒之多）。这意味着仅使用tcp.maxConnections会导致过多的请求排队，等待连接释放。如前面所述，默认情况下，可以排队的请求数为2³²-1。要真正具有熔断（即快速失败）行为，我们还需要设置http.http1MaxPendingRequests限制可以排队的请求数量。它的默认值为1024。有趣的是，如果我们将它的值设置为0，它就会回落到默认值。所以我们必须至少将它设置为1。让我们更新目标规则以仅允许1 个待处理请求：apiVersion: networking.istio.io/v1alpha3 kind: DestinationRule metadata: name: circuit-breaker-sample-server spec: host: circuit-breaker-sample-server trafficPolicy: connectionPool: tcp: maxConnections: 5 http: http1MaxPendingRequests: 1并重启客户端 pod（记得一定需要重启客户端, 否则统计结果会出现偏差）, 并继续观察日志, 类似如下：----------Info---------- Status: 503, Start: 02:56:40, End: 02:56:40, Elapsed Time: 0.005339622497558594 Status: 503, Start: 02:56:40, End: 02:56:40, Elapsed Time: 0.007254838943481445 Status: 503, Start: 02:56:40, End: 02:56:40, Elapsed Time: 0.0044133663177490234 Status: 503, Start: 02:56:40, End: 02:56:40, Elapsed Time: 0.008964776992797852 Status: 200, Start: 02:56:40, End: 02:56:45, Elapsed Time: 5.018309116363525 Status: 200, Start: 02:56:40, End: 02:56:45, Elapsed Time: 5.017424821853638 Status: 200, Start: 02:56:40, End: 02:56:45, Elapsed Time: 5.019804954528809 Status: 200, Start: 02:56:40, End: 02:56:45, Elapsed Time: 5.01643180847168 Status: 200, Start: 02:56:40, End: 02:56:45, Elapsed Time: 5.025975227355957 Status: 200, Start: 02:56:40, End: 02:56:50, Elapsed Time: 10.01716136932373可以看到, 4 个请求立即被限制，5 个请求发送到目标服务，1 个请求排队。这是预期的行为。可以看到客户端 Istio 代理与目标服务中的pod 建立的活跃连接数为5：kubectl exec $(kubectl get pod --selector app=circuit-breaker-sample-client --output jsonpath='{.items[0].metadata.name}') -c istio-proxy -- curl -X POST http://localhost:15000/clusters | grep circuit-breaker-sample-server | grep cx_active outbound|9080||circuit-breaker-sample-server.default.svc.cluster.local::172.20.192.124:9080::cx_active::5场景#2：一个客户端实例和多个目标服务实例现在让我们在有一个客户端实例和多个目标服务实例 pod 的场景中运行测试。首先，我们需要将目标服务部署扩展到多个副本（比如3 个）：kubectl scale deployment/circuit-breaker-sample-server --replicas=3在这里要验证的两个场景分别是：连接限制应用于 pod 级别：目标服务的每个 pod 最多 5 个连接;或者它是否应用于服务级别：无论目标服务中的 pod 数量如何，总共最多 5 个连接;在(1)中，我们应该看不到限制或排队，因为允许的最大连接数为 15（3 个 pod，每个 pod 5 个连接）。由于我们一次只发送 10 个请求，所有请求都应该成功并在大约 5 秒内返回。在(2)中，我们应该看到与之前场景 #1大致相同的行为。让我们再次启动客户端 pod 并监控日志：----------Info---------- Status: 503, Start: 03:06:20, End: 03:06:20, Elapsed Time: 0.011791706085205078 Status: 503, Start: 03:06:20, End: 03:06:20, Elapsed Time: 0.0032286643981933594 Status: 503, Start: 03:06:20, End: 03:06:20, Elapsed Time: 0.012153387069702148 Status: 503, Start: 03:06:20, End: 03:06:20, Elapsed Time: 0.011871814727783203 Status: 200, Start: 03:06:20, End: 03:06:25, Elapsed Time: 5.012892484664917 Status: 200, Start: 03:06:20, End: 03:06:25, Elapsed Time: 5.013102769851685 Status: 200, Start: 03:06:20, End: 03:06:25, Elapsed Time: 5.016939163208008 Status: 200, Start: 03:06:20, End: 03:06:25, Elapsed Time: 5.014261484146118 Status: 200, Start: 03:06:20, End: 03:06:25, Elapsed Time: 5.01246190071106 Status: 200, Start: 03:06:20, End: 03:06:30, Elapsed Time: 10.021712064743042我们仍然看到类似的限制和排队，这意味着增加目标服务的实例数量不会增加客户端的限制。因此, 我们推导出该限制适用于服务级别。运行一段时间之后, 可以看到客户端 Istio 代理与目标服务中的每个 pod 建立的连接数：outbound|9080||circuit-breaker-sample-server.default.svc.cluster.local::172.20.192.124:9080::cx_active::2 outbound|9080||circuit-breaker-sample-server.default.svc.cluster.local::172.20.192.158:9080::cx_active::2 outbound|9080||circuit-breaker-sample-server.default.svc.cluster.local::172.20.192.26:9080::cx_active::2客户端代理与目标服务中的每个 pod 有 2 个活动连接, 不是 5，而是 6。正如 Envoy 和 Istio 文档中所提到的，代理在连接数量方面允许一些回旋余地。场景#3：多个客户端实例和一个目标服务实例在这种情况下，我们有多个客户端实例 pod，而目标服务只有一个 pod。相应地缩放副本：kubectl scale deployment/circuit-breaker-sample-server --replicas=1 kubectl scale deployment/circuit-breaker-sample-client --replicas=3由于所有 Istio 代理都基于本地信息独立运行，无需相互协调，因此对这个测试的期望是每个客户端 pod 都会表现出场景 #1 的行为，即每个 pod 将有 5 个请求被立即发送到目标服务，1 个请求正在排队并受到限制。让我们看一下日志，看看实际发生了什么：Client 1 ----------Info---------- Status: 503, Start: 03:10:40, End: 03:10:40, Elapsed Time: 0.008828878402709961 Status: 503, Start: 03:10:40, End: 03:10:40, Elapsed Time: 0.010806798934936523 Status: 503, Start: 03:10:40, End: 03:10:40, Elapsed Time: 0.012855291366577148 Status: 503, Start: 03:10:40, End: 03:10:40, Elapsed Time: 0.004465818405151367 Status: 503, Start: 03:10:40, End: 03:10:40, Elapsed Time: 0.007823944091796875 Status: 503, Start: 03:10:40, End: 03:10:40, Elapsed Time: 0.06221342086791992 Status: 503, Start: 03:10:40, End: 03:10:40, Elapsed Time: 0.06922149658203125 Status: 503, Start: 03:10:40, End: 03:10:40, Elapsed Time: 0.06859922409057617 Status: 200, Start: 03:10:40, End: 03:10:45, Elapsed Time: 5.015282392501831 Status: 200, Start: 03:10:40, End: 03:10:50, Elapsed Time: 9.378434181213379 Client 2 ----------Info---------- Status: 503, Start: 03:11:00, End: 03:11:00, Elapsed Time: 0.007795810699462891 Status: 503, Start: 03:11:00, End: 03:11:00, Elapsed Time: 0.00595545768737793 Status: 503, Start: 03:11:00, End: 03:11:00, Elapsed Time: 0.013380765914916992 Status: 503, Start: 03:11:00, End: 03:11:00, Elapsed Time: 0.004278898239135742 Status: 503, Start: 03:11:00, End: 03:11:00, Elapsed Time: 0.010999202728271484 Status: 200, Start: 03:11:00, End: 03:11:05, Elapsed Time: 5.015426874160767 Status: 200, Start: 03:11:00, End: 03:11:05, Elapsed Time: 5.0184690952301025 Status: 200, Start: 03:11:00, End: 03:11:05, Elapsed Time: 5.019806146621704 Status: 200, Start: 03:11:00, End: 03:11:05, Elapsed Time: 5.0175628662109375 Status: 200, Start: 03:11:00, End: 03:11:05, Elapsed Time: 5.031521558761597 Client 3 ----------Info---------- Status: 503, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.012019157409667969 Status: 503, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.012546539306640625 Status: 503, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.013760805130004883 Status: 503, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.014089822769165039 Status: 503, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.014792442321777344 Status: 503, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.015463829040527344 Status: 503, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.01661539077758789 Status: 200, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.02904224395751953 Status: 200, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.03912043571472168 Status: 200, Start: 03:13:20, End: 03:13:20, Elapsed Time: 0.06436014175415039结果显示, 每个客户端上的 503 数量增加了。系统仅允许来自所有三个客户端实例pod 的 5 个并发请求。检查客户端代理日志，希望能得到一些线索，并观察到两种不同类型的日志，用于被限制的请求 ( 503 )。其中, 注意到RESPONSE_FLAGS包括了两个值：UO和URX。UO：上游溢出（断路）URX: 请求被拒绝，因为已达到上游重试限制 (HTTP)或最大连接尝试次数 (TCP) 。{"authority":"circuit-breaker-sample-server:9080","bytes_received":"0","bytes_sent":"81","downstream_local_address":"192.168.142.207:9080","downstream_remote_address":"172.20.192.31:44610","duration":"0","istio_policy_status":"-","method":"GET","path":"/hello","protocol":"HTTP/1.1","request_id":"d9d87600-cd01-421f-8a6f-dc0ee0ac8ccd","requested_server_name":"-","response_code":"503","response_flags":"UO","route_name":"default","start_time":"2023-02-28T03:14:00.095Z","trace_id":"-","upstream_cluster":"outbound|9080||circuit-breaker-sample-server.default.svc.cluster.local","upstream_host":"-","upstream_local_address":"-","upstream_service_time":"-","upstream_transport_failure_reason":"-","user_agent":"python-requests/2.21.0","x_forwarded_for":"-"} {"authority":"circuit-breaker-sample-server:9080","bytes_received":"0","bytes_sent":"81","downstream_local_address":"192.168.142.207:9080","downstream_remote_address":"172.20.192.31:43294","duration":"58","istio_policy_status":"-","method":"GET","path":"/hello","protocol":"HTTP/1.1","request_id":"931d080a-3413-4e35-91f4-0c906e7ee565","requested_server_name":"-","response_code":"503","response_flags":"URX","route_name":"default","start_time":"2023-02-28T03:12:20.995Z","trace_id":"-","upstream_cluster":"outbound|9080||circuit-breaker-sample-server.default.svc.cluster.local","upstream_host":"172.20.192.84:9080","upstream_local_address":"172.20.192.31:58742","upstream_service_time":"57","upstream_transport_failure_reason":"-","user_agent":"python-requests/2.21.0","x_forwarded_for":"-"} 带有UO标志的请求由客户端代理在本地进行限制。带有URX标志的请求被目标服务代理拒绝。日志中其他字段的值（例如DURATION、UPSTREAM_HOST 和 UPSTREAM_CLUSTER）也证实了这一点。为了进一步验证，我们还要检查目标服务端的代理日志：{"authority":"circuit-breaker-sample-server:9080","bytes_received":"0","bytes_sent":"81","downstream_local_address":"172.20.192.84:9080","downstream_remote_address":"172.20.192.31:59510","duration":"0","istio_policy_status":"-","method":"GET","path":"/hello","protocol":"HTTP/1.1","request_id":"7684cbb0-8f1c-44bf-b591-40c3deff6b0b","requested_server_name":"outbound_.9080_._.circuit-breaker-sample-server.default.svc.cluster.local","response_code":"503","response_flags":"UO","route_name":"default","start_time":"2023-02-28T03:14:00.095Z","trace_id":"-","upstream_cluster":"inbound|9080||","upstream_host":"-","upstream_local_address":"-","upstream_service_time":"-","upstream_transport_failure_reason":"-","user_agent":"python-requests/2.21.0","x_forwarded_for":"-"} {"authority":"circuit-breaker-sample-server:9080","bytes_received":"0","bytes_sent":"81","downstream_local_address":"172.20.192.84:9080","downstream_remote_address":"172.20.192.31:58218","duration":"0","istio_policy_status":"-","method":"GET","path":"/hello","protocol":"HTTP/1.1","request_id":"2aa351fa-349d-4283-a5ea-dc74ecbdff8c","requested_server_name":"outbound_.9080_._.circuit-breaker-sample-server.default.svc.cluster.local","response_code":"503","response_flags":"UO","route_name":"default","start_time":"2023-02-28T03:12:20.996Z","trace_id":"-","upstream_cluster":"inbound|9080||","upstream_host":"-","upstream_local_address":"-","upstream_service_time":"-","upstream_transport_failure_reason":"-","user_agent":"python-requests/2.21.0","x_forwarded_for":"-"}正如预期的那样，这里有 503 响应码，这也是导致客户端代理上有 "response_code":"503"以及"response_flags":"URX"。总而言之, 客户端代理根据它们的连接限制（每个 pod最多 5 个连接）发送请求——排队或限制（使用UO 响应标志）多余的请求。所有三个客户端代理在批处理开始时最多可以发送 15 个并发请求。但是，其中只有 5 个成功，因为目标服务代理也在使用相同的配置（最多 5 个连接）限制。目标服务代理将仅接受 5 个请求并限制其余请求，这些请求在客户端代理日志中带有URX响应标志。上述所发生情况的直观描述类似如下：场景#4：多个客户端实例和多个目标服务实例最后一个也可能是最常见的场景，我们有多个客户端实例pod和多个目标服务实例 pod 。当我们增加目标服务副本时，我们应该会看到请求的成功率整体增加，因为每个目标代理可以允许 5 个并发请求。如果我们将副本数增加到 2，我们应该会看到所有 3 个客户端代理在一个批次中生成的 30 个请求中有 10 个请求成功。我们仍然会观察到客户端和目标服务代理上的限制。如果我们将副本增加到 3，我们应该看到 15 个成功的请求。如果我们将数量增加到 4，我们应该仍然只能看到 15 个成功的请求。为什么？这是因为无论目标服务有多少个副本，客户端代理上的限制都适用于整个目标服务。因此，无论有多少个副本，每个客户端代理最多可以向目标服务发出 5 个并发请求。总结在客户端：每个客户端代理独立应用该限制。如果限制为 100，则每个客户端代理在应用本地限制之前可以有 100 个未完成的请求。如果有 N 个客户端调用目标服务，则总共最多可以有 100*N 个未完成的请求。客户端代理的限制是针对整个目标服务，而不是针对目标服务的单个副本。即使目标服务有 200 个活动 pod, 限流仍然会是100。在目标服务端：每个目标服务代理也适用该限制。如果该服务有 50 个活动的 pod，则在应用限流并返回 503 之前，每个 pod 最多可以有 100 个来自客户端代理的未完成请求。如果对服务网格 ASM 感兴趣或者对上述内容有任何疑问，欢迎钉钉扫描下方二维码或搜索群号（30421250）加入服务网格ASM 用户交流群，一起探索服务网格技术。

文章

监控 · Kubernetes · 网络协议 · Java · API · Python · Perl · 容器

2023-02-28

做 SQL 性能优化真是让人干瞪眼

很多大数据计算都是用SQL实现的，跑得慢时就要去优化SQL，但常常碰到让人干瞪眼的情况。比如，存储过程中有三条大概形如这样的语句执行得很慢： select a,b,sum(x) from T group by a,b where …; select c,d,max(y) from T group by c,d where …; select a,c,avg(y),min(z) from T group by a,c where …;这里的T是个有数亿行的巨大表，要分别按三种方式分组，分组的结果集都不大。分组运算要遍历数据表，这三句SQL就要把这个大表遍历三次，对数亿行数据遍历一次的时间就不短，何况三遍。这种分组运算中，相对于遍历硬盘的时间，CPU计算时间几乎可以忽略。如果可以在一次遍历中把多种分组汇总都计算出来，虽然CPU计算量并没有变少，但能大幅减少硬盘读取数据量，就能成倍提速了。如果SQL支持类似这样的语法：from T --数据来自T表 select a,b,sum(x) group by a,b where … --遍历中的第一种分组 select c,d,max(y) group by c,d where … --遍历中的第二种分组 select a,c,avg(y),min(z) group by a,c where …; --遍历中的第三种分组能一次返回多个结果集，那就可以大幅提高性能了。可惜， SQL没有这种语法，写不出这样的语句，只能用个变通的办法，就是用group a,b,c,d的写法先算出更细致的分组结果集，但要先存成一个临时表，才能进一步用SQL计算出目标结果。SQL大致如下： create table T\_temp as select a,b,c,d, sum(case when … then x else 0 end) sumx, max(case when … then y else null end) maxy, sum(case when … then y else 0 end) sumy, count(case when … then 1 else null end) county, min(case when … then z else null end) minz group by a,b,c,d; select a,b,sum(sumx) from T\_temp group by a,b where …; select c,d,max(maxy) from T\_temp group by c,d where …; select a,c,sum(sumy)/sum(county),min(minz) from T\_temp group by a,c where …; 这样只要遍历一次了，但要把不同的WHERE条件转到前面的case when里，代码复杂很多，也会加大计算量。而且，计算临时表时分组字段的个数变得很多，结果集就有可能很大，最后还对这个临时表做多次遍历，计算性能也快不了。大结果集分组计算还要硬盘缓存，本身性能也很差。还可以用存储过程的数据库游标把数据一条一条fetch出来计算，但这要全自己实现一遍WHERE和GROUP的动作了，写起来太繁琐不说，数据库游标遍历数据的性能只会更差！只能干瞪眼！TopN运算同样会遇到这种无奈。举个例子，用Oracle的SQL写top5大致是这样的： select \* from (select x from T order by x desc) where rownum<=5表T有10亿条数据，从SQL语句来看，是将全部数据大排序后取出前5名，剩下的排序结果就没用了！大排序成本很高，数据量很大内存装不下，会出现多次硬盘数据倒换，计算性能会非常差！避免大排序并不难，在内存中保持一个5条记录的小集合，遍历数据时，将已经计算过的数据前5名保存在这个小集合中，取到的新数据如果比当前的第5名大，则插入进去并丢掉现在的第5名，如果比当前的第5名要小，则不做动作。这样做，只要对10亿条数据遍历一次即可，而且内存占用很小，运算性能会大幅提升。这种算法本质上是把TopN也看作与求和、计数一样的聚合运算了，只不过返回的是集合而不是单值。SQL要是能写成这样,就能避免大排序了： select top(x,5) from T然而非常遗憾，SQL没有显式的集合数据类型，聚合函数只能返回单值，写不出这种语句！不过好在全集的TopN比较简单，虽然SQL写成那样，数据库却通常会在工程上做优化，采用上述方法而避免大排序。所以Oracle算那条SQL并不慢。但是，如果TopN的情况复杂了，用到子查询中或者和JOIN混到一起的时候，优化引擎通常就不管用了。比如要在分组后计算每组的TopN，用SQL写出来都有点困难。Oracle的SQL写出来是这样： select \* from (select y,x,row\_number() over (partition by y order by x desc) rn from T) where rn<=5这时候，数据库的优化引擎就晕了，不会再采用上面说的把TopN理解成聚合运算的办法。只能去做排序了，结果运算速度陡降！假如SQL的分组TopN能这样写： select y,top(x,5) from T group by y把top看成和sum一样的聚合函数，这不仅更易读，而且也很容易高速运算。可惜，不行。还是干瞪眼！关联计算也是很常见的情况。以订单和多个表关联后做过滤计算为例，SQL大体是这个样子：select o.oid,o.orderdate,o.amount from orders o left join city ci on o.cityid = ci.cityid left join shipper sh on o.shid=sh.shid left join employee e on o.eid=e.eid left join supplier su on o.suid=su.suid where ci.state='New York' and e.title='manager' and ... 订单表有几千万数据，城市、运货商、雇员、供应商等表数据量都不大。过滤条件字段可能会来自于这些表，而且是前端传参数到后台的，会动态变化。SQL一般采用HASH JOIN算法实现这些关联，要计算 HASH 值并做比较。每次只能解析一个JOIN，有N个JOIN要执行N遍动作，每次关联后都需要保持中间结果供下一轮使用，计算过程复杂，数据也会被遍历多次，计算性能不好。通常，这些关联的代码表都很小，可以先读入内存。如果将订单表中的各个关联字段预先做序号化处理，比如将雇员编号字段值转换为对应雇员表记录的序号。那么计算时，就可以用雇员编号字段值（也就是雇员表序号），直接取内存中雇员表对应位置的记录，性能比HASH JOIN快很多，而且只需将订单表遍历一次即可，速度提升会非常明显！也就是能把SQL写成下面的样子：select o.oid,o.orderdate,o.amount from orders o left join city c on o.cid = c.# --订单表的城市编号通过序号#关联城市表 left join shipper sh on o.shid=sh.# --订单表运货商号通过序号#关联运货商表 left join employee e on o.eid=e.# --订单表的雇员编号通过序号#关联雇员表 left join supplier su on o.suid=su.#--订单表供应商号通过序号#关联供应商表 where ci.state='New York' and e.title='manager' and ... 可惜的是，SQL 使用了无序集合概念，即使这些编号已经序号化了，数据库也无法利用这个特点，不能在对应的关联表这些无序集合上使用序号快速定位的机制，只能使用索引查找，而且数据库并不知道编号被序号化了，仍然会去计算 HASH 值和比对，性能还是很差！有好办法也实施不了，只能再次干瞪眼！还有高并发帐户查询，这个运算倒是很简单：select id,amt,tdate,… from T where id='10100' and tdate>= to\_date('2021-01-10','yyyy-MM-dd') and tdate<to_date('2021-01-25','yyyy-mm-dd') and="" …="" <p="">在T表的几亿条历史数据中，快速找到某个帐户的几条到几千条明细，SQL写出来并不复杂，难点是大并发时响应速度要达到秒级甚至更快。为了提高查询响应速度，一般都会对 T 表的 id 字段建索引：create index index_T_1 on T(id)在数据库中，用索引查找单个帐户的速度很快，但并发很多时就会明显变慢。原因还是上面提到的SQL无序理论基础，总数据量很大，无法全读入内存，而数据库不能保证同一帐户的数据在物理上是连续存放的。硬盘有最小读取单位，在读不连续数据时，会取出很多无关内容，查询就会变慢。高并发访问的每个查询都慢一点，总体性能就会很差了。在非常重视体验的当下，谁敢让用户等待十秒以上？！容易想到的办法是，把几亿数据预先按照帐户排序，保证同一帐户的数据连续存储，查询时从硬盘上读出的数据块几乎都是目标值，性能就会得到大幅提升。但是，采用SQL体系的关系数据库并没有这个意识，不会强制保证数据存储的物理次序！这个问题不是SQL语法造成的，但也和SQL的理论基础相关，在关系数据库中还是没法实现这些算法。那咋办？只能干瞪眼吗？不能再用SQL和关系数据库了，要使用别的计算引擎。开源的集算器SPL基于创新的理论基础，支持更多的数据类型和运算，能够描述上述场景中的新算法。用简单便捷的SPL写代码，在短时间内能大幅提高计算性能！上面这些问题用SPL写出来的代码样例如下：一次遍历计算多种分组 AB1=file("T.ctx").open().cursor(a,b,c,d,x,y,z 2cursor A1=A2.select(…).groups(a,b;sum(x))3 //定义遍历中的第一种过滤、分组4cursor=A4.select(…).groups(c,d;max(y))5 //定义遍历中的第二种过滤、分组6cursor=A6.select(…).groupx(a,c;avg(y),min(z))7 //定义遍历中的第三种过滤、分组8…//定义结束，开始计算三种方式的过滤、分组用聚合的方式计算Top5全集Top5（多线程并行计算）| | A | | --- | --- | |1 |=file("T.ctx").open()||2 |=A1.cursor@m(x).total(top(-5,x),top(5,x))||3 |//top(-5,x) 计算出 x 最大的前 5 名，top(5,x) 是 x 最小的前 5 名。|分组Top5（多线程并行计算）| | A | | --- | --- | |1 |=file("T.ctx").open()||2 |=A1.cursor@m(x,y).groups(y;top(-5,x),top(5,x))|用序号做关联的SPL代码：系统初始化| | A | | --- | --- | |1 |>env(city,file("city.btx").import@b()),env(employee,file("employee.btx").import@b()),...||2 |//系统初始化时，几个小表读入内存|查询| | A | | --- | --- | |1 |=file("orders.ctx").open().cursor(cid,eid,…).switch(cid,city:#;eid,employee:#;…)||2| =A1.select(cid.state=="New York" && eid.title=="manager"…)||3 |//先序号关联，再引用关联表字段写过滤条件|高并发帐户查询的SPL代码：数据预处理，有序存储| | A | B| | --- | --- | --- | |1| =file("T-original.ctx").open().cursor(id,tdate,amt,…)|||2| =A1.sortx(id)| =file("T.ctx")||3| =B2.create@r(#id,tdate,amt,…).append@i(A2)|||4| =B2.open().index(index_id;id)| ||5| //将原数据排序后，另存为新表，并为帐号建立索引||帐户查询| | A | | --- | --- | |1| =T.icursor(;id==10100 && tdate>=date("2021-01-10") && tdate<date("2021-01-25") && …,index_id).fetch()||2| //查询代码非常简单|除了这些简单例子，SPL还能实现更多高性能算法，比如有序归并实现订单和明细之间的关联、预关联技术实现多维分析中的多层维表关联、位存储技术实现上千个标签统计、布尔集合技术实现多个枚举值过滤条件的查询提速、时序分组技术实现复杂的漏斗分析等等。正在为SQL性能优化头疼的小伙伴们，可以和我们一起探讨：http://www.raqsoft.com.cn/wx/Query-run-batch-ad.htmlSPL资料SPL官网SPL下载SPL源代码

文章

SQL · 存储 · 数据采集 · 并行计算 · 算法 · Oracle · 关系型数据库 · 大数据 · 数据库 · 索引

2023-02-13

CVPR2021 | 视觉推理解释框架VRX：用结构化视觉概念作为解释网络推理逻辑的「语言」

大家好，我是Charmve。今天分享的一篇文章来自葛云皓，本文主要介绍了被 CVPR 2021 录用的文章《A Peek Into the Reasoning of Neural Networks: Interpreting with Structural Visual Concepts》。本文提出了一个视觉推理解释框架 (VRX: Visual Reasoning eXplanation), 将人们容易理解的、high-level 的结构化的视觉概念作为「语言」，通过回答为什么是 A，为什么不是 B 解释神经网络的推理逻辑。VRX 还可以利用解释对网络进行诊断，进一步提升原网络的性能。这项工作对神经网络推理逻辑的可解释性进行了探究：区分于大多数现有 xAI 方法通过可视化输入图像和输出结果之间的相关性对网络进行解释，该研究提出用结构化的视觉概念 (Structural Visual Concept) 对神经网络决策背后的推理逻辑和因果关系进行解释，通过解答网络决策中「为什么是 A？为什么不是 B？」的问题，用人们更容易理解的 high-level 视觉概念和视觉概念之间的结构和空间关系解释神经网络的推理逻辑，并将其作为一种直接指导来提升被解释网络的性能。论文地址：https://arxiv.org/pdf/2105.00290.pdf项目主页：http://ilab.usc.edu/andy/vrxGitHub 地址：https://github.com/gyhandy/Visual-Reasoning-eXplanation一作主页：https://gyhandy.github.io/下图概括了这篇文章要做的任务：为了解释原网络决策背后的推理逻辑，该研究回答了如下问题：为什么是消防车？为什么不是救护车？又为什么不是校车？该研究用结构化的视觉概念图(Structural Concept Graph) 作为解释的语言，其中概念图的节点 (node) 代表视觉概念(visual concept)，边 (edge) 代表视觉概念之间的结构和空间关系，点和边的颜色代表其对该类最终决策的贡献度（冷色：正向或暖色：负向）：（1）为什么是消防车？从视觉概念角度，所有检测到的四个与消防车最相关的视觉概念（保险杠，消防车头，车轮，救援架）都对最终消防车的决策有正向贡献；从视觉概念的空间结构关系角度，四个概念之间的空间关系也都对决策有正向贡献，这说明视觉概念和他们之间的关系都像一辆消防车。（2）为什么不是校车？从视觉概念角度：从图中检测到的与校车视觉概念最接近的四个部分及其相关结构和空间关系都对校车的决策起到负向贡献（否定该图是校车的决策）尤其是概念 1 和概念 2，与真正的校车概念最不相符。（3）为什么不是救护车？也同样可以得到相似的人们容易理解的，逻辑上的，视觉概念角度的解释。下面将详细介绍工作的具体内容。研究动机在深度学习日益蓬勃发展的今天，深度神经网络不透明的决策导致的安全事故和隐患也越来越多，神经网络的可解释性对于人们如何更加信任、安全、可靠的使用他们至关重要。近年来有越来越多关注可解释性的研究，例如：pixel-level 的方法 (CAM[1]Grad-CAM[2]等) 通过可视化输入图像和输出结果之间的相关性解释网络的决策，为理解神经网络决策依据找到了线索；concept-level 的方法 (TCAV[3]ACE[4]等) 可以找到给定类别重要的视觉概念。然而，这些方法是否局限于解释相对 low-level 的相关性？是否有更加方便人们理解的更直观的 high-level 的解释方法？我们是否可以揭示神经网络内在的推理逻辑和因果关系？逻辑解释能否作为线索进一步帮助提高原网络的性能？为了回答这些问题，该研究探究如何模拟和解释神经网络的推理逻辑，提出用结构化的视觉概念对神经网络决策背后的推理逻辑和因果关系进行解释，通过解答网络决策中「为什么是 A，为什么不是 B？」的问题，用人们更容易理解的 high-level 视觉概念和视觉概念之间的关系解释神经网络的推理逻辑，并将其作为指导来提升原网络的性能。方法详述该研究提出的视觉推理解释框架（VRX：Visual Reasoning eXplaination Framework）包括三个主要部分：(1)视觉概念提取器 (VCE: Visual Concept Extractor) 用来提取特定类别相关的重要视觉概念，并将图像表示为结构化的视觉概念图 (SCG: Structural Concept Graph)；(2)概念图推理网络 (GRN: Graph Reasoning Network) 以视觉概念图为输入，通过知识蒸馏和迁移来模拟原网络的决策过程；(3)可视化决策解释器 (VDI: Visual Decision Interpreter) 用来解释原网络决策背后的推理逻辑和因果关系。接下来对每个部分进行详细解释。视觉概念提取器和结构化的视觉概念该研究首先介绍了什么是视觉概念 (Visual Concept)，简单来说视觉概念展示了给定神经网络对不同类别的理解，同时人们也更容易接受符合直觉的概念级别的解释：以下图警车为例，在给定的神经网络「眼里」，警车 Top2 重要的视觉概念可视化为最右边绿色圈中的 patch (看起来像轮子和驾驶室侧面)。ACE[4] 中作者对视觉概念进行了定义：类别相关的视觉概念是像素点的集合 (group of pixels) 并满足以下三个要求：(1)有意义 (Meaningfulness)：即视觉概念需要具有语义上的涵义，单个的像素就没有语义涵义，所以需要是像素点的集合，比如图片 patch。(2)一致性 (Coherency)：同一视觉概念在不同图片中的表现应该相似，不同视觉概念之间应该不同。(3)重要性 (Importance)：如果一个视觉概念的存在对于该类样本的真实预测是必要的，那么它对于该类的预测就是「重要的」。下图描述了 ACE 中对给定网络、给定类别的视觉概念的提取过程：以警车为例，首先 (a) 用多分辨率的分割算法对图片进行分割得到 patch（这里的分割采用的是 SLIC[5]，一种基于规则的分割算法，选择该方法是对于其速度和效果的综合考虑）; 然后(b) 将分割得到的 patch resize 为统一大小，通过给定网络将 patch 转化为向量，并在向量空间做聚类; 最后(c) 利用 Testing with Concept Activation Vectors (TCAV)[3] 得到每个聚类的潜在视觉概念对警车类别的重要性分数，并剔除 outlier，留下对警车类别来说最重要的 top 视觉概念。但研究发现，ACE 提取视觉概念的效果非常依赖用于提取视觉概念的图片的质量，一般每一类选取 50~100 张左右的图片用于提取视觉概念，如果图片有一些 bias 或者不是很具有代表性，就会导致很多提取的视觉概念落在背景区域，比如下图（左），这些视觉概念并不能代表网络学习了该类 1000 张图像（ImageNet）以后对该类（救护车）的理解。为了解决这个问题，该研究提出使用自顶向下的梯度注意力 (Attention Map) 对视觉概念提取区域进行约束， 因为 Grad-CAM 的 attention map 可以高亮显示对网络决策重要的区域（多为前景），这样可以帮助剔除掉提取视觉概念的图片中与类别无关的背景部分，使得提取的视觉概念更能代表原网络对该类的理解，如下图右。以吉普车类别为例，下图总结了视觉概念提取器提取视觉概念的步骤。提取出类别相关的视觉概念后，研究者认为视觉概念之间有潜在的空间结构关系，这种空间关系对类别表达至关重要：比如我们并不能说，只要能在图像中检测到吉普车的四个最重要的视觉概念就代表一定是吉普车，他们之间的空间关系是相对确定的，例如轮子不能在车顶上方。我们人类做决策也是相似的：我们认为这是一辆吉普车，不仅关键的特征（视觉概念）符合认知，特征之间的空间关系同样会影响我们的推理和最终决策。因此该研究认为结构化的视觉概念才是更符合人们直觉的，解释神经网络推理决策的重要「语言」。后续的模拟并解释神经网络的推理决策过程也是基于此展开的。结构化视觉概念的一种表达便是结构概念图 (SCG: Structural Concept Graph)，graph 中不同颜色的点代表不同重要性 (Top k) 的视觉概念，边代表视觉概念之间的空间关系。如下图吉普车和斑马的例子，我们可以把任意图片表达为对应类别的结构概念图。注意：目前是 image-level 的 SCG (I-SCG)，后续会用基于 learning 的方法，学习到 class-level 的 SCG(c-SCG)。概念图推理网络有了结构概念图作为人们容易理解的解释神经网络的「语言」，接下来可以用这种「语言」解释神经网络决策背后的推理逻辑。ACE[4]的作者为了验证提取到的视觉概念对神经网络决策的重要性，实验验证发现：如果只保留输入图片中表达重要视觉概念的像素（mask 掉与重要视觉概念无关的区域），神经网络能保留原本 80% 以上的准确率。因此一个比较直接的想法便是：我们能不能追踪并可视化神经网络决策过程中重要视觉概念相关特征的流动，这样我们便能找到最终决策与重要视觉概念之间的关系，从而对决策进行解释。以下图为例对于一个经典的由卷积层和全连接层构成的神经网络，我们用不同的颜色代表吉普车最重要的四个视觉概念，在卷积层，我们可以根据结构不变性追踪每个视觉概念对应的 representation feature。但是全连接层中，所有特征耦合到了一起，使得追踪变得困难。研究者分析这是由于神经网络结构上信息流动不够透明和难以解耦导致的，与此同时研究者想到另一种解决办法：如果可以用另一个结构比较解耦的，信息流动更加透明的模型 B，全方位模拟原始神经网络 A 的推理和决策，是不是就可以通过解释 B 的推理逻辑来解释 A 呢？为此研究者提出概念图推理网络 (GRN: Graph Reasoning Network)，以结构概念图为输入，通过知识蒸馏和迁移来模拟原网络的决策过程（如下图）。下图解释了概念图推理网络的训练过程：对于输入图片，该研究首先构建对于每个感兴趣类别的结构概念图（即先将图像进行分割，然后在所有 patch 中分别检测每个感兴趣类别的重要视觉概念：从下图中检测到 2 个消防车的视觉概念（黄色圆圈），2 个老爷车的视觉概念 ... 4 个吉普车的视觉概念），这些检测到的视觉概念组成相应类别的结构概念图，表示对其决策的假设（该图是消防车吗？是老爷车吗 ... 是吉普车吗？）很多类别我们只能检测到部分视觉概念，检测不到的视觉概念用 dummy node 来表示（黑色 node）。然后概念图推理网络利用图卷积，对每一个结构概念图进行 representation，学习其视觉概念及其之间的关系对最终决策的影响。最后该研究把所有点和边 concatenate 为一个向量， 通过非常简单的 MLP 输出对所有感兴趣类别的决策向量，并用知识蒸馏的方法使得概念图推理网络与原网络的决策一致。为了提升模拟的鲁棒性，该研究还用 mask out 视觉概念添加扰动的方法使得概念图推理网络与被解释的原网络在面对扰动时决策一致。（详细训练过程和公式推导请见原始 paper）。训练中，所有类别的结构概念图共享一套图卷积的参数，但是每个类别在消息传递（message passing ） 中有专属的注意力权重参数 eji，类别专属的注意力权重参数是为了学习每个类别独特的视觉概念之间的空间和依赖关系，一方面可以解释并可视化概念图推理网络学习到的每个类别视觉概念之间潜在的关系（下图），另一方面为最终推理过程的解释提供了支持。下图是用学习到的消防车的 eji 筛选出重要的视觉概念之间的关系。边的 eji 值越大，代表点 j 对点 i 的贡献越大；从右边的 sum 可以看到消防车的视觉概念 1 和 2 对其他的视觉概念贡献最大，这也意味着他们是对消防车来说最有区分度的视觉概念。可视化决策解释器 (VDI: Visual Decision Interpreter)训练好的概念图推理网络便是原网络的 representation，基于图卷积神经网络的概念图推理网络具有信息传递透明且容易追踪的特点，为了用结构概念图对推理过程进行解释，该研究提出了基于梯度的贡献度分配算法，为每个参与决策的点（视觉概念）和边（概念之间的关系）计算其对于特定决策的贡献值，贡献值的高低代表了其肯定还是否定了该决策。最后决策解释器可视化了对原网络输出的解释并回答为什么是吉普车？为什么不是其他类别？（颜色代表肯定：冷色，或否定：暖色）如下图右：（1）为什么是吉普车？从视觉概念角度，所有检测到的四个与吉普车最相关的视觉概念（前灯，挡风玻璃，后窗，车轮）都对最终吉普车的决策有正向贡献（深蓝或浅蓝）；从视觉概念的空间结构关系角度，四个概念之间的空间关系也都对决策有正向贡献，这说明视觉概念和他们之间的关系都像一辆吉普车。（2）为什么不是消防车？从视觉概念角度：从图中检测到的与消防车视觉概念最接近的四个部分及其相关结构和空间关系都对消防车的决策起到负向贡献（否定该图是消防车的决策）尤其是概念 1 和概念 2，与真正的消防车概念最不相符。（3）为什么不是老爷车等？也同样可以得到相似的人们容易理解的、逻辑上的、视觉概念角度的解释。实验和结果视觉推理解释 (VRX) 与原网络之间逻辑一致性实验第一个实验目的是验证视觉推理解释框架 (VRX) 做出的推理解释与原网络的逻辑是一致的。如下图，原网络 Xception 错把一张消防车分类成了救护车，VRX 给出解释（如左图）：为什么不是消防车？因为从图像中检测到的消防车的视觉概念 3 和 4 都对消防车的决策起到负向贡献即否定该决策。为什么是救护车？因为检测到的救护车的视觉概念 3 和 4 都对救护车的决策起到正向贡献，即肯定该决策。即使所有消防车视觉概念之间的空间关系（边）相对救护车的空间关系更加合理，但是综合来看，Xception 还是做出了救护车的决策。为了验证解释的合理性以及与原网络决策逻辑的一致性。该研究做了两个实验 :（1）研究人员把原图中检测到的对消防车决策起负向作用的消防车的概念 3 (车轮) 替换为另外一张消防车图片中的更合理的概念 3 (右图第一行)，然后让 Xception 对新的图片再次分类，发现错误被纠正了。此外，该研究也做了对比试验：如果用一张随机的消防车的 patch 去替换概念 3，或者用另外一张消防车的同样合理的概念 1 和 2 替换原始的概念 1 和 2，Xception 都无法纠正错误。所以研究者认为 VRX 对 Xception 推理的解释符合原网络的逻辑。（2）研究人员把原图中检测到的对救护车的决策起正向作用的救护车的概念 3 mask 掉，发现 Xception 对新图片的预测结果有纠正的趋势（消防车概率增大，救护车概率减小）。对比实验发现如果随机删除 patch 则不会有纠正的效果。上述验证实验一共在 119 张 Xception 错分的图片中实施，研究者用 VRX 对错误原因背后推理逻辑的解释作为修改建议，通过视觉概念的替换和删除，原网络超过 95% 的错分可以被正确的纠正（如下表 1）。VRX 对视觉和结构解释的敏感性实验VRX 可以从视觉概念（点）和视觉概念之间的空间关系（边）两个角度为决策提供解释。通过添加扰动，分别预设输入图像的视觉层面不合理和空间结构的不合理，探究 VRX 的解释对视觉和结构的敏感性。给定一张 Xception 正确预测的救护车图片，VRX 给出了解释。（a）如果把一个救护车相对合理的视觉概念 2（对救护车决策起到正向贡献）替换为一个相对不合理的视觉概念 2（在另一张救护车图像中对决策起到负向贡献），VRX 对新图片决策的解释可以正确捕捉到不合理的部分：被替换的概念 2。（b）如果把视觉概念 4（车轮）移动到一个不合理的位置（挡风玻璃上方），VRX 对新图片决策的解释可以正确捕捉到不合理的部分：视觉概念 4 和其他视觉概念之间的空间关系。由此研究者认为 VRX 可以准确的定位视觉和结构的不合理，并给出准确的解释。VRX 根据解释对原网络进行诊断并提升原网络的表现之前的实验展示了 VRX 对原网络的解释可以帮助原网络纠正对图片的错误分类，接下来的实验中，VRX 将利用可解释性对原网络进行诊断，发现原网络训练中存在的问题（比如训练数据的 bias），从而提出针对性修改建议，进而提升原网络的表现。如下图，研究者用 Resnet-18 训练了一个三种车辆的分类器 ，(a) 训练数据有 pose 的 bias，所有的公共汽车都是 pose 1，所有坦克都是 pose 2，所有军用汽车都是 pose 3；但测试数据没有 pose bias，即所有车辆都有全部的 pose 1，2 和 3。(b) 测试发现分类器的准确率较低，该研究用 VRX 对模型进行诊断，发现大部分错分的图像其实都能找到正确的视觉概念，错误原因是因为概念之间的关系否定了正确的决策，导致错分。以下图 (b) 为例，一辆军用汽车被错分为坦克，解释为什么不是军用汽车的时候发现是军用汽车视觉概念之间的空间关系否定了该图是军用汽车的决策，而从图中检测到的坦克的视觉概念虽然较差，但是空间关系支持是坦克的决策，综合以上导致了错分。所以 VRX 诊断给出的建议是增加视觉概念之间空间关系的多样性和鲁棒性。根据诊断建议，最直接的实现方法便是增加不同 pose 的图片，研究者做了接下来的验证实验：setting 1：为每类增加 150 张不同于原始数据集 pose 的图片；setting 2：为每类增加 150 张与原始数据集 pose 相同的图片（对照组）。研究者用新的数据集分别重新训练了 Resnet-18 分类器并测试数据集准确率 (结果如下表 2)。VRX 根据解释性的诊断帮助提升了原始模型的效果。VRX 对原模型的诊断可以总结模型的不同问题并提出相应提升建议。下图是 VRX 对在 ImageNet 上预训练的 Xception 其中六类的诊断，VRX 将其错误总结为三种类别，并对每种错误的修改提出建议。（细节请参考原 paper）。总结和展望总结来说，这项工作在神经网络的可解释性方向做了进一步探索：解释神经网络决策背后的推理逻辑。该研究提出了一个视觉推理解释框架 (VRX: Visual Reasoning eXplanation), 将人们容易理解的、high-level 的结构化的视觉概念作为“语言”，通过回答为什么是 A，为什么不是 B 解释神经网络的推理逻辑。VRX 还可以利用解释对网络进行诊断，进一步提升原网络的性能。研究者相信这是朝着更透明、更安全、更可信的 AI 方向迈出的小但是重要的一步。论文讲解视频：https://www.bilibili.com/video/BV1tb4y1C7yY/GitHub开源中文学习数字媒介——结合图示、代码及HTML，只需一个浏览器就可以学习，实战促进理论学习。在线电子书 https://charmve.github.io/computer-vision-in-action/GitHub - Charmve/computer-vision-in-action: 《计算机视觉实战演练：算法与应用》中文电子书、源码、读者交流社区（持续更新中 ...）参考文献1. CAM, Bolei Zhou , et al. 2016 https://openaccess.thecvf.com/content_cvpr_2016/papers/Zhou_Learning_Deep_Features_CVPR_2016_paper.pdf2. Grad-CAM, RR Selvaraju , et al. 2017 https://arxiv.org/pdf/1610.02391.pdf3. TCAV, Been Kim, et al. 2017; https://arxiv.org/pdf/1711.11279.pdf4. ACE, Ghorbani, et al. 2019 https://arxiv.org/pdf/1902.03129.pdf5. SLIC, Radhakrishna Achanta, et al. 2011 https://core.ac.uk/download/pdf

文章

机器学习/深度学习 · 编解码 · 人工智能 · 数据可视化 · 算法 · 安全 · 数据挖掘 · 计算机视觉

2023-02-10

Python3，自从我掌握了Doscoart库，我对艺术的成就越来越高。

1、引言小屌丝：鱼哥，最近在忙啥？小鱼：咱俩陌生了？小屌丝：何出此言？小鱼：你说的话又嘛意思呢？小屌丝：我的意思， 最近看你这整理各种资料，貌似很忙的样子？小鱼：我平时不也这么忙嘛小屌丝：鱼哥， 还能正常唠嗑嘛？小鱼：我又没说不能唠嗑。小屌丝：鱼哥，行… 非常行…小鱼：男人，怎么能不行!小屌丝：…小鱼：~ ~小屌丝：discoart模块知道吗？小鱼：貌似， 大概，可能，或许，知道。小屌丝：太好了， 那能不能给我讲一讲呢？小鱼：然后呢？小屌丝：老地方~小鱼：又是老地方，整的我都不好意思了。小屌丝：这都是小事，我主要就想让你多放松放松…小鱼：停，停， 打住~ 别说多了， 我们来聊discoart。小屌丝：别着急啊，小鱼：能不着急吗，你看，这都几点了， 再晚一会，就…小屌丝：昂…2、代码实战2.1 模块介绍说起 discoart 可能大部分都不太了解。但是，说到艺术库，可能你就有些印象了。这里，我也引用官网对discoart的解析，让你对它有个初步的了解，如下：DiscoArt is an elegant way of creating compelling Disco Diffusion[*] artworks for generative artists, AI enthusiasts and hard-core developers. DiscoArt has a modern & professional API with a beautiful codebase, ensuring high usability and maintainability. It introduces handy features such as result recovery and persistence, gRPC/HTTP serving w/o TLS, post-analysis, easing the integration to larger cross-modal or multi-modal applications. 这里，我也简答的用汉语描述一下，即：DiscoArt是一种优雅的方式，可以为生成艺术家，AI爱好者和铁杆开发人员创建引人注目的Disco Diffusion艺术品。 DiscoArt拥有现代和专业的API，具有漂亮的代码库，确保了高可用性和可维护性。它引入了方便的功能，例如结果恢复和持久性，没有TLS的gRPC / HTTP服务，后期分析，简化与更大的跨模态或多模态应用程序的集成。简答一句话概括：DiscoArt就是为了艺术而生的。2.2 模块安装涉及到第三方库，肯定就需要安装老规矩，pip 安装pip install discoart然后就是等待着安装。其它安装方式，直接看这两篇:《Python3，选择Python自动安装第三方库，从此跟pip说拜拜！！》《Python3：我低调的只用一行代码，就导入Python所有库！》安装的样子，如下：这里提示一下：按照官网的要求， discoart的使用，必须依托于：Python 3.7+ 和 CUDA 的 PyTorch 。2.3 代码示例2.3.1 创建默认图片这里直接使用discoart的 create方法即可代码示例：# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 使用默认参数创建图片 ''' from discoart import create ca = create()效果展示2.3.2 设置参数创建图片代码示例# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 设置参数创建图片 ''' from discoart import create #设置参数 ca = create( text_prompts='A painting of sea cliffs in a tumultuous storm, Trending on ArtStation.', init_image='https://d2vyhzeko0lke5.cloudfront.net/xxxx7e77b72f0.png', skip_steps=100, )效果展示2.3.3 查看设置参数如果你忘记参数，也没关系，直接用cheatsheet 查询即可代码展示# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 查看设置参数 ''' from discoart import cheatsheet #设置参数 sha = cheatsheet()2.3.4 查看配置如果要查看文档配置， 可以使用show_config：代码示例# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 查看文档配置 ''' from discorat import show_config # 展示第一个项目运行的配置 show_config(da) # 参考第四个项目的运行配置 show_config(da[3]) #查看discoartID show_config('discoart-xxxxfbf288')2.3.5 保存配置如果要保存文档配置， 可以使用save_config：代码展示# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 保存文档配置 ''' from discoart import save_config #保存第一次运行的配置 save_config(da, 'my.yml') #保存第四次运行的配置 save_config(da[3], 'my.yml')2.3.6 加载配置有了查看和保存，当然也可以直接加载配置文件了， 这里，使用load_config即可代码示例# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 加载文档配置 ''' from discoart import create, load_config #加载配置文件 config = load_config('my.yml') create(**config)2.3.7 导出配置文件为了便于后期的管理使用，同样可以直接导出配置文件为SVG映像，使用 save_config_svg方法：代码示例# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 导出配置文件为SVG映像 ''' from discoart.config import save_config_svg #直接保存为svg映像 save_config_svg(da)这里也展示一下， 保存的svg映像2.3.8 生成Python代码更神奇的功能，就是可以直接从配置中生成可运行的Python代码，使用export_python方法：代码示例# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 生成可运行的Python代码 ''' from discoart.config import export_python export_python(da)2.3.9 调用文档如果你觉得自己配置太繁琐， 那可以直接使用DocumentArray作为初始状态运行。代码示例# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 调用DocumentArray作为初始状态，运行 ''' from discoart import create from docarray import DocumentArray da = DocumentArray.pull('discoart-32xxx') create( init_document=da[0], cut_ic_pow=0.5, tv_scale=600, cut_overview='[12]*1000', cut_innercut='[12]*1000', use_secondary_model=False, )当然， 如果你只想从已有的 DocArray ID 初始化， 那也不是不可能，# -*- coding:utf-8 -*- # @Time : 2023-02-12 # @Author : Carl_DJ ''' 实现功能： 从已有的 DocArray ID 初始化 ''' from discoart import create create(init_document='discoart-320xxxx')3、总结看到这里， Doscart库的介绍就完成了。按照流程， 我们来回顾一下今天都分享了啥内容：创建默认图片；设置参数创建图片；查看设置参数；查看文档配置；报错文档配置；加载文档配置；导出配置文件生成Python代码；调用文档；你看， Doscoart也没有想想的那么难嘛， 常用的功能，也就差不多这么多。所以， 只要我们把一个库从头到尾的捋一遍， 其实是很容易掌握的。最后，唠叨一句：我是奕然：CSDN 博客专家；阿里云专家博主；51CTO 博客专家；51认证讲师；金牌面试官&面试培训师；关注我，带你学习更多更有趣的Python知识。

文章

并行计算 · PyTorch · 算法框架/工具 · Python

2023-02-20

Golang 中那些隐秘的角落

我们真的有用好 recover() 吗？在一次系统报警调查中，发现某个组件 panic 且没有恢复运行，panic 的日志为是 "fatal error: concurrent map writes"，当时只能手动重启该组件。查看其源码时发现，panic 位置对应的函数中，已经存在 recover() 函数，抽象得到的源码逻辑如下所示：package main import ( "fmt" ) func concurrentMapWrite() { defer func() { if err := recover(); err != nil { fmt.Printf("Panic occurred due to %+v, Recovered in f", err) } }() m := map[int]int{} idx := 0 for { go func() { m[idx] = 1 }() idx++ } } func main() { concurrentMapWrite() } 当时初步判断 recover 并没有捕获到 "fatal error: concurrent map writes"，为了验证自己的猜想，进行了如下的一系列调查。在 defer 中使用 recover()Golang 程序运行不符合预期时往往会通过“错误”以及“异常”来反馈给用户。前者是代码逻辑出现错误时返回，是编程者意料之中的错误，不会破坏程序的运行；后者往往是代码中出现了不可预期的错误，导致程序无法继续运行下去，如果不加以处理，就会导致程序异常退出，这是十分危险的。为了提高程序的稳健性，我们需要依赖 Golang 的 recover 以及 defer 机制来保证程序在发生异常后能够继续维持运行，避免程序意外退出，保证业务的稳定运行。defer 关键字中包含的内容会在其所在函数返回之前执行；recover 函数用于将 goroutine 从异常场景下恢复，只有在 defer 中调用才会生效。其使用方式如下如下：func div(x, y int) int { return x / y } func f() { defer func() { if err := recover(); err != nil { fmt.Printf("Panic occurred due to %+v, Recovered in f", err) } }() fmt.Println(div(1, 0)) }上述 defer...recover 相当于 java 中的 try/catch 关键字，能够保证程序能够不被异常中断执行，我们知道使用 try...catch 可以捕获所有类型的异常，只要 catch 后跟所有异常的基类 Exception 即可；然后Golang为什么却不是这样呢？不可恢复的 panic不同于 try...catch，在 Golang 中并不是所有异常都能够被 recover 捕获到：当异常是通过 runtime.panic() 抛出时，能够被 recover 方法捕获；当异常是通过 runtime.throw() 或者 runtime.fatal() 抛出时，不能够被 recover 方法捕获。在上述实际场景中遇到的 “concurrent map writes” 异常就是通过 runtime.fatal() 抛出来的，具体源码(runtime/map.go)：// Like mapaccess, but allocates a slot for the key if it is not present in the map. func mapassign(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer { if h == nil { panic(plainError("assignment to entry in nil map")) } // 省略 ... if h.flags&hashWriting != 0 { fatal("concurrent map writes") } hash := t.hasher(key, uintptr(h.hash0)) // Set hashWriting after calling t.hasher, since t.hasher may panic, // in which case we have not actually done a write. h.flags ^= hashWriting if h.buckets == nil { h.buckets = newobject(t.bucket) // newarray(t.bucket, 1) } again: // 省略 ... bucketloop: // 省略 ... done: if h.flags&hashWriting == 0 { fatal("concurrent map writes") } h.flags &^= hashWriting if t.indirectelem() { elem = *((*unsafe.Pointer)(elem)) } return elem }map 通过标志位 h.flags 来检查 map 是否存并发写情况，如果存在，则调用 fatal 方法，此时错误为 "fatal error"，会强制退出程序，详情见 fatal 源码：// fatal triggers a fatal error that dumps a stack trace and exits. // // fatal is equivalent to throw, but is used when user code is expected to be // at fault for the failure, such as racing map writes. // // fatal does not include runtime frames, system goroutines, or frame metadata // (fp, sp, pc) in the stack trace unless GOTRACEBACK=system or higher. // //go:nosplit func fatal(s string) { // Everything fatal does should be recursively nosplit so it // can be called even when it's unsafe to grow the stack. systemstack(func() { print("fatal error: ", s, "\n") }) fatalthrow(throwTypeUser) } // fatalthrow implements an unrecoverable runtime throw. It freezes the // system, prints stack traces starting from its caller, and terminates the // process. // //go:nosplit func fatalthrow(t throwType) { pc := getcallerpc() sp := getcallersp() gp := getg() if gp.m.throwing == throwTypeNone { gp.m.throwing = t } // Switch to the system stack to avoid any stack growth, which may make // things worse if the runtime is in a bad state. systemstack(func() { startpanic_m() if dopanic_m(gp, pc, sp) { // crash uses a decent amount of nosplit stack and we're already // low on stack in throw, so crash on the system stack (unlike // fatalpanic). crash() } exit(2) }) *(*int)(nil) = 0 // not reached }从 fatal 方法的注释中可知，该方法等同于 throw 方法，但是只会抛出用户层面的异常，系统层面的异常由 runtime.throw 抛出。fatal 方法中又调用了 fatalthrow 方法， 该方法的注释中明确表示："fatalthrow implements an unrecoverable runtime throw"，因此通过这个方法抛出的异常均属于不可恢复异常。"concurrent map writes" 之所以被视为不可恢复异常，是因为 Golang 检测到数据竞争时，map 内部的结构已经被破坏了，继续运行可能会产生不可预期的结果，因此会强制结束程序。以下罗列了一些其他不可恢复的异常种类：Out of memoryConcurrent map writesStack memory exhaustionAttempting to launch a nil function as a goroutineAll goroutines are asleep - deadlockThread limit exhaustion参考：[1] https://go-review.googlesource.com/c/go/+/390421[2] https://github.com/golang/go/blob/master/src/runtime/map.go#L578[3] https://stackoverflow.com/questions/57486620/are-all-runtime-errors-recoverable-in-go[4] https://www.jianshu.com/p/15c459c85141[5] https://www.zhihu.com/question/305845656/answer/728440889切片扩容有哪些坑？在开发过程中，将切片作为参数传递到函数中，然后在函数中修改切片内容，对应的修改预期一定能够同步更新到实参中，然而实际开发测试中发现，有的场景符合预期，有的却不符合预期。若在函数中对该切片进行扩容且扩容后的切片大小不超过其原始容量，此时修改切片中已有的元素，则修改会同步到实参切片中，而扩容不会同步到实参切片中；若在函数中对该切片进行扩容且扩容后的切片大小超过其原始容量，则修改不会同步到实参切片中，同时扩容不会同步到实参切片中，示例代码如下：package main import ( "fmt" ) func appendSliceWithinCap(s []string) { s = append(s, "two") s[0] = "appendSliceWithinCap" } func appendSliceOverCap(s []string) { s = append(s, "two") s = append(s, "three") s[0] = "appendSliceOverCap" } func main() { fmt.Println("hello main") s := make([]string, 1, 2) s[0] = "one" fmt.Println(s) // ["one"] appendSliceWithinCap(s) fmt.Println(s) // ["appendSliceWithinCap"] appendSliceOverCap(s) fmt.Println(s) // ["appendSliceWithinCap"] } 切片扩容机理函数中对切片参数中已有元素的更新会影响实参切片自身维护了一个指针属性，用于指向它底层数组中的某些元素的集合，其结构体如下所示：type slice struct { array unsafe.Pointer // 指向底层数组的指针 len int // 切片的长度 cap int // 切片的容量 }Golang 官方文档声明：函数参数传参只有值传递一种方式。值传递方式会在调用函数时将实际参数拷贝一份传递到函数中，slice 参数被传递到函数中时，其 array、len 以及 cap 都被复制了一份，因此函数中 slice 和实参 slice 是共享了底层 slice 数组的，函数中对 slice 中已有元素的修改会同步到实参 slice 中。切片的扩容策略切片可以通过 append 函数以追加元素的方式进行动态扩容，扩容的元素会存储在切片已有的存储空间中，然而切片的存储空间上限由切片容量决定，当扩容的元素数量超过切片容量时，切片必须对底层数组进行扩容才能容纳这些元素，我们通过 go/src/runtime/slice.go 中的 growslice 方法来解析下此时 Golang（1.19.2+） 扩容切片的策略：// growslice allocates new backing store for a slice. // // arguments: // // oldPtr = pointer to the slice's backing array // newLen = new length (= oldLen + num) // oldCap = original slice's capacity. // num = number of elements being added // et = element type // // return values: // // newPtr = pointer to the new backing store // newLen = same value as the argument // newCap = capacity of the new backing store // // Requires that uint(newLen) > uint(oldCap). // Assumes the original slice length is newLen - num // // A new backing store is allocated with space for at least newLen elements. // Existing entries [0, oldLen) are copied over to the new backing store. // Added entries [oldLen, newLen) are not initialized by growslice // (although for pointer-containing element types, they are zeroed). They // must be initialized by the caller. // Trailing entries [newLen, newCap) are zeroed. // // growslice's odd calling convention makes the generated code that calls // this function simpler. In particular, it accepts and returns the // new length so that the old length is not live (does not need to be // spilled/restored) and the new length is returned (also does not need // to be spilled/restored). func growslice(oldPtr unsafe.Pointer, newLen, oldCap, num int, et *_type) slice { oldLen := newLen - num // 省略 ... newcap := oldCap doublecap := newcap + newcap if newLen > doublecap { newcap = newLen } else { const threshold = 256 if oldCap < threshold { newcap = doublecap } else { // Check 0 < newcap to detect overflow // and prevent an infinite loop. for 0 < newcap && newcap < newLen { // Transition from growing 2x for small slices // to growing 1.25x for large slices. This formula // gives a smooth-ish transition between the two. newcap += (newcap + 3*threshold) / 4 } // Set newcap to the requested cap when // the newcap calculation overflowed. if newcap <= 0 { newcap = newLen } } } // 省略 ... // The check of overflow in addition to capmem > maxAlloc is needed // to prevent an overflow which can be used to trigger a segfault // on 32bit architectures with this example program: // // type T [1<<27 + 1]int64 // // var d T // var s []T // // func main() { // s = append(s, d, d, d, d) // print(len(s), "\n") // } if overflow || capmem > maxAlloc { panic(errorString("growslice: len out of range")) } var p unsafe.Pointer if et.ptrdata == 0 { p = mallocgc(capmem, nil, false) // The append() that calls growslice is going to overwrite from oldLen to newLen. // Only clear the part that will not be overwritten. // The reflect_growslice() that calls growslice will manually clear // the region not cleared here. memclrNoHeapPointers(add(p, newlenmem), capmem-newlenmem) } else { // Note: can't use rawmem (which avoids zeroing of memory), because then GC can scan uninitialized memory. p = mallocgc(capmem, et, true) if lenmem > 0 && writeBarrier.enabled { // Only shade the pointers in oldPtr since we know the destination slice p // only contains nil pointers because it has been cleared during alloc. bulkBarrierPreWriteSrcOnly(uintptr(p), uintptr(oldPtr), lenmem-et.size+et.ptrdata) } } memmove(p, oldPtr, lenmem) return slice{p, newLen, newcap} }当扩容后的元素总数超过切片容量大小，其扩容策略如下：threshold = 256若扩容后的元素数量超过两倍原始容量，则直接将扩容后元素数量赋值给新容量，否则执行如下若原容量小于 threshold，则将原始容量的两倍赋值给新容量，否则执行如下在原始容量基础上，每次增加 (原始容量 + threshold * 3)/ 4，直到其不小于扩容后的元素数量扩容策略完成，得到新容量值后，会基于该值申请内存，然后将原数组中的数据以及扩容的数据拷贝到新内存中，此时完成切片的动态扩容，其公式如下：由上可知，当函数对形参 slice 进行扩容且扩容后的元素数量超过原始切片容量时，底层数组会迁移到另片内存区域，因此函数中对形参 slice 已有元素的更新无法影响到实参 slice。参考：[1] https://www.liangtian.me/post/go-slice/[2] https://juejin.cn/post/6888117219213967368[3]https://github.com/golang/go/blob/4c61e079c087052355c137ab8fcd9abf8728e50a/src/runtime/slice.goContext 是如何影响 grpc 通信超时控制的上述场景是我在实际开发应用过程中抽象出来的 grpc 通信过程，这也是一个较为通用的过程，client 端将带有超时时间的 context 传递到 server 端，server 端在超时时间内需要完成请求处理并返回响应给 client 端，若超过超时请求时间，那么链接将会断开，client 端将不会收到任何响应。然而在实际开发应用中，发现即便 server 端的 context 超时了，但是其请求响应仍会偶发性地传递到 client 端，导致我们的一个功能出现了不可预期的情况，为了用代码描述对应的交互流程，我在这里放了简化后的示例代码，描述了当时的交互逻辑。grpc 超时传递流程在 Golang grpc 通信过程中，超时信息会在不同通信端进行传递的，传递的介质是 Http2 Request Frame。grpc client 在发送请求之前，会将信息封装在不同的的 Frame 中，例如 Data Frame 用来存放请求的 response payload；Header Frame 用户存在一些跨 goroutine 传递的数据，例如路径信息。而超时信息就存放在 Header Frame 中，其源码如下所示：// NewStream 方法的调用链路：grpc.Invoke -> invoke -> sendRequest -> NewStream // NewStream creates a stream and register it into the transport as "active" // streams. func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) { // 省略 ... // HPACK encodes various headers. Note that once WriteField(...) is // called, the corresponding headers/continuation frame has to be sent // because hpack.Encoder is stateful. t.hBuf.Reset() t.hEnc.WriteField(hpack.HeaderField{Name: ":method", Value: "POST"}) t.hEnc.WriteField(hpack.HeaderField{Name: ":scheme", Value: t.scheme}) t.hEnc.WriteField(hpack.HeaderField{Name: ":path", Value: callHdr.Method}) t.hEnc.WriteField(hpack.HeaderField{Name: ":authority", Value: callHdr.Host}) t.hEnc.WriteField(hpack.HeaderField{Name: "content-type", Value: "application/grpc"}) t.hEnc.WriteField(hpack.HeaderField{Name: "user-agent", Value: t.userAgent}) t.hEnc.WriteField(hpack.HeaderField{Name: "te", Value: "trailers"}) if callHdr.SendCompress != "" { t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-encoding", Value: callHdr.SendCompress}) } if dl, ok := ctx.Deadline(); ok { // Send out timeout regardless its value. The server can detect timeout context by itself. timeout := dl.Sub(time.Now()) t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-timeout", Value: encodeTimeout(timeout)}) } // 省略 ... }client server 端在收到超时信息后，将 grpc-timeout 字段从 Header 中取出，基于该超时信息新建一个 context 实例，其源码如下所示：// processHeaderField 方法调用链：grpc.Server -> handleRawConn -> serveNewHTTP2Transport -> serveStreams -> HandleStreams -> operateHeaders -> processHeaderField // operateHeader takes action on the decoded headers. func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream)) (close bool) { buf := newRecvBuffer() s := &Stream{ id: frame.Header().StreamID, st: t, buf: buf, fc: &inFlow{limit: initialWindowSize}, } var state decodeState for _, hf := range frame.Fields { state.processHeaderField(hf) } // 省略 ... s.recvCompress = state.encoding if state.timeoutSet { s.ctx, s.cancel = context.WithTimeout(context.TODO(), state.timeout) } else { s.ctx, s.cancel = context.WithCancel(context.TODO()) } // 省略 ... } func (d *decodeState) processHeaderField(f hpack.HeaderField) { switch f.Name { // 省略 ... case "grpc-timeout": d.timeoutSet = true var err error d.timeout, err = decodeTimeout(f.Value) if err != nil { d.setErr(streamErrorf(codes.Internal, "transport: malformed time-out: %v", err)) return } // 省略 ... } }在 grpc client 端，会去不断检查 context.Done() 来判断 context 是否超时，若超时，则会断开链接。然而，也会存在 context timeout races 的情况，例如，client 端 context 已经超时，但是此时下一轮检查还未开始，同时 server 端恰好返回了响应信息，此时虽然 client 端 context 超时了，但是仍然会接收到 server 端的响应并处理；更普遍的情况是 select { case <- ctx; ...; case <- response; ...}，这就会导致有 50% 的概率未检测到 context 超时，详情请参考我之前在 grpc-go 中提的 issue。确保 grpc 响应超时错误在我之前经历的错误场景中， server 端 context 出现超时，并返回响应给 client 端，此时 client 端预期应该也会超时并断开链接，但实际是会成功接收到 client 端的响应，由于处理逻辑的问题，当时的响应并不包含超时错误，因此 client 端在接收到请求后会重新发送一次请求，重新发送完成后，才检测到 context 超时，最终断开链接，导致了错误的出现。因此，在应用过程中，需要在 server 端 context timeout 时，保证返回的 response 中的错误信息是 grpc.DeadlineExceeded，让 client 端也感知到 timeout 的发生，避免不必要逻辑的发生。参考：[1] https://github.com/grpc/grpc-go[2] https://github.com/grpc/grpc-go/issues/5206#issuecomment-1058564271[3] https://xiaomi-info.github.io/2019/12/30/grpc-deadline/

文章

存储 · Java · Serverless · Go

2022-12-07

基于阿里云SLS日志服务的数据安全管理

基于阿里云SLS日志服务的数据安全管理伴随云原生的发展，以应用为核心的可观测监控越发得到广泛应用。在应用的可观测性中，日志的收集及内容监控是可观测性监控的一个重要内容，然而日志的监控数据，并不同于某项指标的测量结果数据，其往往都会包含业务或与用户相关的敏感性信息，因此，在落实日志的收集和监控的同时，我们也要注意日志数据的安全管理。阿里云SLS阿里云日志服务SLS是云原生观测与分析平台，其可为Log、Metric、Trace等数据提供大规模、低成本、实时的存储、归档、监控等一系列的数据服务，其包括一站式提供数据采集、加工、查询与分析、可视化、告警、消费与投递等功能，可满足研发、运维、运营、安全等场景的数据管理能力。阿里云SLS功能特性如图所示，其功能特点如下：数据采集日志服务提供50多种数据接入方案，支持采集服务器与应用相关的日志、时序数据和链路数据。支持基于Logtail 采集物联网设备、阿里云产品日志、移动端等数据。支持Logstash、Flume、Beats、FluentD、Telegraph等开源软件中数据的接入。支持通过HTTP、HTTPS、Syslog、Kafka、Prometheus等标准协议接入数据。查询与分析支持上下文查询、日志聚类、LiveTail、重建索引等功能。支持标准的SQL 92语法。数据加工日志服务提供数据加工功能，用于数据的规整、富化、流转、脱敏和过滤。日志脱敏可对数据中包含的密码、手机号、地址等敏感信息进行脱敏。消费与投递日志服务提供消费与投递功能，支持通过SDK、API实时消费数据；支持通过控制台将数据实时投递至OSS、MaxCompute等阿里云产品中。数据可视化日志服务支持可视化展示查询和分析结果。提供表格、线图、柱状图等多种统计图表支持直接对接Grafana、DataV等第三方图表内容告警日志服务提供告警监控、降噪、事务管理、通知分派的智能运维平台支持通过告警策略对所接收到的告警进行路由分派、抑制、去重、静默、合并等操作，然后发送给通知（行动）管理系统。支持通过Webhook方式接收外部监控系统中的告警消息（例如Grafana告警、Prometheus告警），并完成告警管理、告警通知等操作。日志审计覆盖基础（ActionTrail、容器服务Kubernetes版）、存储（OSS、NAS）、网络（SLB、API网关）、数据库（关系型数据库RDS、云原生分布式数据库PolarDB-X、PolarDB MySQL云原生数据库）、安全（WAF、DDoS防护、云防火墙、云安全中心）等云产品。阿里云的数据安全方案阿里云SLS 被定位一个集中化的数据管理平台，如上所述，其涵盖了数据从采集到审计归档的全生命周期内容。对于数据的安全，可包含两个方面的内容，分别是数据脱敏和用户安全。数据脱敏数据脱敏，是指数据的内容在被用户查询过程中，用户所看到的内容不会出现违反安全法规限定的内容，同时也无法根据不同的内容碎片组成违反安全法规限定的内容。阿里云SLS 提供了在采集端和服务端数据加工两种方案来满足这一需求。采集端控制这是一种常见的数据脱敏方法，通常日志在采集后经过相关策略的处理，再把脱敏的数据传送到SLS服务端进行存储，这种方法相对较为安全，但是由于采集端的多样性，若采集端软件不同，各采集端配置上则会较为分散，也不方便管理。此处以阿里云日志采集端logtail的配置为例来了解采集端数据脱敏的配置方法。Logtail 的日志处理流程如下图所示：基于logtail 实现数据脱敏有字段丢弃、数据脱敏、字段加密、数据编码与解码等方法，logtail 脱敏配置样例如下：{ "processors":[ { "type":"processor_drop", "detail": { "DropKeys": ["aaa1","aaa2"] } }, { "type" : "processor_desensitize", "detail" : { "SourceKey" : "password", "Method" : "const", "Match" : "full", "ReplaceString": "********" } }, { "type" : "processor_desensitize", "detail" : { "SourceKey" : "content", "Method" : "md5", "Match" : "regex", "RegexBegin": "'password':'", "RegexContent": "[^']*" } }, { "type" : "processor_encrypt", "detail" : { "SourceKey" : ["field1","field2"], "EncryptionParameters" : { "Key": "18b2eda21fbca92d6d3442e951c15208d0ae50d0a3e5d9e45bbaea0f5121a09d519c90a676133f36d48a0288e2a928bbe83f67d9f7ef5ba5508cbe8f63b3ecb79b2394b132467424395bdb40c97f3929fad2ae627117367d9e425f4776b229af0d934e0f95a04087286bf58dda46fa16aae77db19567709ead743631ba48fe80998bdecc5ea23d66d37ce821828ab840660738c696eda24c222e1cc64929b13b391655b9056cd0729bbc44231e98e3301fbe0b401c4628a682cbf1a3b9c9b5d7ddffe196034f2a66e866e37ffa64f81235026d32c72744c5bdc831fdfc5624c53eaae44134b169de665bd3d7fb5eeeab3853c301aea39dd5358766e004702fdc6f5ae79edc975936c50c80c8d4853ead9e8b52531cbef3747cde2308f7bfe2df25b65e738035b485b58fa23772814b396ee787a9aa8c6826921d4af7e450715fae0f77ed5684631912c198430a19c50fe22ac0e2eda4e7ae91d271f2560078aa6ff88b5ffec53de505d04be1876174bebb9fdb8fbbd2faaed65c8e37669d731e9404bb52c54af5e02d4d6816534bdf72504d97581a0049aa041f773e8b4ebc21a021c728215d13d1f28fd9b9fe0a3623b19c4e03e3c056116a765f9ef594b63815eb747aade065772504dee454116a60fa29dd1b3b74ef4477e74aa1d8aee6f5c36e9f30fbe51b5561c8c3ad0d31d0ec1726d4e3166986f526800c9f358652df" } } }, { "type" : "processor_base64_encoding", "detail" : { "SourceKey" : "content", "NewKey" : "content_base64" } }, { "type" : "processor_base64_decoding", "detail" : { "SourceKey" : "content_base64", "NewKey" : "content" } }, { "type" : "processor_base64_md5", "detail" : { "SourceKey" : "content", "NewKey" : "content" } } ] }数据加工数据加工则是在数据存储到SLS 服务端后，通过相关的数据处理规则实现数据内容的脱敏，该方法易于数据的集中处理，缺点是未脱敏的数据仍会被存储到在SLS服务端，需要做好相关的用户权限控制来解决这一弊端。数据加工原理数据加工的原理如图所示，其过程是将源存储的日志中数据同步到目标的日志库中，在同步的过程中按照数据加工规则，将加工后的数据存储到目标日志库中。数据加工规则引擎使用SLS DSL提供的内置函数编写加工规则，每个函数可以看做一个加工步骤，规则引擎按步骤顺序执行函数。配置样例# 过滤所有健康检测的日志 e_if(e_search("actuator/health"), e_drop()) # 过滤掉无用字段 e_drop_fields("field1","field2") # 替换掉手机号 e_set( "message", regex_replace( v("message"), r"((13[0-9]|14[01456879]|15[0-35-9]|16[2567]|17[0-8]|18[0-9]|19[0-35-9])\d{8})", replace=r"[MASKED_PHONE]", ), ) e_set( "log", regex_replace( v("log"), r"(?i)(\suser\s).*", replace=r"\1[MASKED_NAME]" ), )性能优化根据加工原理，数据加工任务的总体速度取决于源Shard的数量、用户配置的规则逻辑和规则复杂度。一般可以按照每Shard处理1MB/s（压缩前）流量规划，也就是大约85 GB每天每Shard规划。例如：源Logstore的数据写入速度是每天1 TB，那么需要分裂源Logstore的Shard数量为1024GB/85=12个。成本优化根据加工原理和性能指南，官方推荐节约成本的方案是使用一个或多个Logstore快速接入数据，然后使用数据加工进行快速分发，并根据实际需求设置不同目标Logstore的存储时间以及索引配置，进而降低使用成本。举例某应用的日志，每天的写入量为100 GB，存储30天并建立全文索引，那么日志服务的成本大约是2248元/每月。官方推荐使用如下加工方案实现成本的节约。* 构建接入源Logstore，存储3天不建立索引。 * 构建目标Logstore1，用于存储操作日志与错误日志，存储30天并建立索引。 * 构建目标logstore2，用于存储一般性日志，存储7天并建立索引。这种情况下您的成本约为1672元/每月，与加工之前相比大约可以节省25%的成本。用户安全用户安全，是指数据使用者是在授权下使用数据，其所有的操作，是在权限控制下的，且不会在操作过程中，对数据产生安全风险。基于这一需求，我们可以通过阿里云提供的权限控制服务实现这一需求。阿里云的RAM 中的相关概念如下:RAM用户是一个身份实体，它通常代表您的组织中需要访问云资源的人员或应用程序。用户组多个可具有相同策略的RAM用户集合。RAM角色是一种虚拟用户，没有确定的身份认证密钥，需要被一个受信的实体用户关联才能正常使用。其可实现同一实体用户的多身份切换，方便权限策略的管理。权限策略用于描述一组权限集，RAM 支持两种类型的权限策略：由阿里云管理的系统策略和由客户管理的自定义策略。系统策略，统一由阿里云创建，只能使用而不能修改。自定义策略则可更具实际的情况自主创建、更新和删除，相对更为灵活。RAM相关概念关系如图所示，RAM用户、用户组及RAM角色都可以直接与权限策略关联，权限策略则可针对阿里云各服务产品的读、写和列表权限实现控制，为增强权限控制的能力，其还可以对每个产品下的资源，即用户的具体云产品的实例实现更细粒度的管理。同时，阿里云的RAM 策略增加了条件控制，对RAM用户的访问方式实现在时间、源IP、安全通道(https)、多因素认证等方式进行了识别，增加了更多的安全管控方法。策略脚本结构及语法自定义策略可以通过可视化编辑器及脚本编辑器创建，由于阿里云产品更新的较快，可视化编辑器的可选项与实际可用权限会存在不同步的情况，建议使用脚本编辑器创建。RAM中使用权限策略描述授权的具体内容，权限策略由效果（Effect）、操作（Action）、资源（Resource）、条件（Condition）和授权主体（Principal）等基本元素组成。权限策略结构包括：版本号(Version)和授权语句列表(Statement)。每条授权语句包括授权效果（Effect）、操作（Action）、资源（Resource）以及条件（Condition，可选项）。策略脚本样例{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:GetConfig", "log:GetConsumerGroupCheckPoint", "log:GetDashboard" ], "Resource": [], "Condition": {} } ] }策略脚本结构<version_block> = "Version" : ("1") // Statement 是个列表，可以添加多个授权语句 <statement_block> = "Statement" : [ <statement>, <statement>, ... ] // 每个Statement 由effect、action、resource、condition 组成。 <statement> = { <effect_block>, <action_block>, <resource_block>, <condition_block?> } // effect 的可选项为"Allow" 或 "Deny" <effect_block> = "Effect" : ("Allow" | "Deny") // action 是各产品的操作描述符 <action_block> = "Action" : ("*" | [<action_string>, <action_string>, ...]) // resource 是各产品具体的用户实例 <resource_block> = "Resource" : ("*" | [<resource_string>, <resource_string>, ...]) // condition 是用户访问方式的识别 <condition_block> = "Condition" : { <condition_type_string> : { <condition_key_string> : [<condition_value>, <condition_value>, ...], <condition_key_string> : [<condition_value>, <condition_value>, ...], ... }, <condition_type_string> : { <condition_key_string> : [<condition_value>, <condition_value>, ...], <condition_key_string> : [<condition_value>, <condition_value>, ...], ... }, ... } <condition_key_string> = ("acs:CurrentTime"|"acs:SecureTransport"|"acs:SourceIp"|"acs:MFAPresent") <condition_value> = ("String" | "Number" | "Boolean" | "Date and time" | "IP address")用户角色的设定基于实际的工作需求，建议由如下四类用户，并根据用户组和RAM角色对每个用户实现更细粒度的管控。用户说明管理超管账户可管理所有的阿里云资源基于审计实现管理控制权限管理员用户的权限设定仅有权限控制产品的操作权限运维人员云产品资源配置仅有被授权产品可工作的最小权限使用者业务范围使用者仅有被授权产品可工作的最小权限权限策略判断流程参考文档https://help.aliyun.com/product/28958.htmlhttps://help.aliyun.com/product/28625.html

文章

存储 · 运维 · 监控 · 安全 · Cloud Native · 数据可视化 · 关系型数据库 · 数据安全/隐私保护 · 对象存储 · 索引

2023-03-20

得物从0到1自研客服IM系统的技术实践之路

本文由得物技术王卫强分享，为了更好的阅读体验，有较多的内容修订和排版优化。一、引言客服IM的核心业务其实就是在线沟通，客服IM的好处是使得客服与用户通过实时沟通的方式可以在最短的时间内帮助用户解决问题。为了快速支撑公司业务发展需求，我们客服IM在发展初期是基于第三方的云IM SDK进行二次开发而来。虽然提升了项目进展，但同时也埋下了问题定位困难、特殊功能实现成本高等隐患。随着公司业务的高速发展，客服对IM聊天的性能和体验都有了更高的要求，在第三方云IM SDK消息通信上逐渐遇到了技术瓶颈。为解决租用第三方云IM SDK接入带来的潜在隐患、提升IM的稳定性和高扩展性，自研一套可控、稳定、灵活的IM系统已是迫在眉睫了。本篇文章将基于工程实践，分享我们从0到1自研一套客服IM系统时在各种关键技术点上的设计思路和实践方法。注：为了简化内容，本文分享的技术栈主要是以Web客服端为主。相关文章：1）从零到卓越：京东客服即时通讯系统的技术架构演进历程；2）瓜子IM智能客服系统的数据架构设计（整理自现场演讲，有配套PPT）。学习交流：- 移动端IM开发入门文章：《新手入门一篇就够：从零开发移动端IM》- 开源IM框架源码：https://github.com/JackJiang2011/MobileIMSDK（备用地址点此）（本文已同步发布于：http://www.52im.net/thread-4153-1-1.html）二、业务场景客服与用户在聊天的过程中，直观上就是客服在输入文案，然后通过网络发送给用户。但是IM聊天SDK该如何设计才能使客服在发送消息过程中感知不到卡顿？这一点是非常关键的，要避免卡顿就要设计合理的发送策略以及避免大量JS脚本执行。举个客服与用户聊天的例子：1）客服发送了“客服小冰为您服务”这个文案，通过业务侧调用SDK的接口，传入到SDK；2）再将该数据存储到数据池中，序列化后把这个数据对象data传递给socket接口，通过网络通道发送到网关；3）网关侧接收到消息后，再反序列化，传递到数据池中进行处理，组装成业务可识别的model，推送到业务侧使用。针对第1）点，SDK会先创建消息体，即把这个字符串封装成一个自定义的结构体model。其聊天流程如下图所示：从上图中可以清晰的看出一条消息发送和接收的完整流程链路。如果IM的SDK设计不合理，发送消息和接收消息流程出现了卡顿，将直接影响用户的体验。三、自研框架架构图概览下图是我们的自研IM系统架构原理图：我们整体的技术改造主要是两个方面：1）对消息链路的抽象改造：主要是消息数据存储和消息排序的重构；2）业务接入侧的抽象改造：主要是将业务逻辑和SDK源码进行解耦，做到代码分层更加的清晰。下面我们将针对主要的技术点进行详细地总结和分享。四、消息链路发布/订阅实现在IM SDK自研开发过程中，如何解耦框架代码和业务代码，做到灵活的消息监听，前期调研之后使用了RxJS。这里简单介绍几个RxJS的核心概念：1）Observable（可观察对象）：表示一个可调用的未来值或事件的集合；2）Observer（观察者）：监听由Observable提供的值；3）Subscription （订阅）：表示 Observable 的执行。注：Subscription 有一个重要的方法，即 unsubscribe，它不需要任何参数，只是用来清理由 Subscription 占用的资源主要用于取消 Observable 的执行。SDK底层在接收到数据后需要同步到业务侧，之前的做法是通过监听方式实现，这种方式不具备取消订阅的能力，维护成本相对较高。而使用RxJS可以清晰的梳理出数据流向，通过发布订阅的方式实现数据的通信。RxJS在发布订阅的实现流程如下：从上图可以看到消息处理的整个流向非常清晰，框架底层接收消息，订阅者消费消息。五、消息框架的分层结构概览在我们整个自研的IM消息通信框架中，主要结构分成三层：1）网络层；2）数据链路层；3）应用层。具体如下图所示：接下来我将详细分享各层的设计和实现思路。六、消息框架的分层实现：网络层网络层作为消息发送的最底层，负责TCP的连接、消息发送和接收。网络协议我们选择的是TCP协议。我们为什么没有选择UDP呢？因为UDP是无连接的、不够安全、无法提供可靠传输的服务，通过TCP连接传送的数据可以无差别、不丢失、不重复且按序到达。PS：尺有所短、寸有所长，TCP和UDP的优劣应该客观看待，感兴趣可以深入学习下面的文章：《快速理解TCP和UDP的差异》《一泡尿的时间，快速搞懂TCP和UDP的区别》《快速理解为什么说UDP有时比TCP更有优势》《深入地理解UDP协议并用好它》《如何让不可靠的UDP变的可靠？》我们整个IM SDK的通信方式采用的是 WebSocket + JSON、grpc + protobuf。（如果你对WebSocket和Protobuf不熟悉，可以详细学习《WebSocket从入门到精通，半小时就够！》、《Protobuf从入门到精通，一篇就够！》）首先我们要做的就是建立Websocket连接：代码层面我们会先创建一个Connection的抽象类，主要处理网络连接相关配置、超时后重新连接的补偿实现，和一些继承类需要实现的抽象方法。如上述代码所示：核心在处理超时重连，传统的重试策略是每隔一段时间重试一次，由于是固定的时间间隔重试，重试时又会有大量的请求在同一时刻涌入，会不断地造成限流。（这里使用了指数退避的方式，指数退避是一种通过反馈，成倍地降低某个过程的速率，以逐渐找到合适速率的算法，可根据时隙和重试尝试次数来决定延迟重试。）其实现算法大致如下：Websocket连接我们是通过继承Connect类实现的：至此：网络层连接就已完成了，相对比较简单，都是一些socket api的封装，核心的点在用指数退避算法实现消息发送失败重连接。七、消息框架的分层实现：数据链路层数据链路层是IM SDK的核心层，主要涉及到用户信息、聊天消息、数据池等等，我们来一步步对每个模块进行分析。首先梳理一下客服在登录到用户进线发送消息和接收消息的全过程。过程有如下几个阶段：7.1、协议类型消息协议类型非常重要，是消息发送的基石。初始化协议数据体，可以用于后续各种消息、事件的发送。在IM自研的SDK通信协议类型主要有如下几种： 具体解释一下：1）Hi：发送客户端基础信息，告诉server当前client的版本、设备类型、语言等信息；2）Login: 登录，token验证，获取或创建当前用户topic信息；3）Sub: 订阅topic或更新topic数据；4）Leave: 取消订阅，解绑之前的订阅关系；5）Pub: 发送数据消息给指定topic的订阅者；6）Get: 获取topic的metadata信息，例如：获取订阅者列表、历史数据等；7）Set: 更新topic的metadata信息，例如：删除消息或删除topic；8）Del: 用于删除操作，包括删除消息、删除订阅关系、删除topic等；9）Note: client发送通知给topic的订阅者，例如消息已收到，消息已读，当前正在输入等；10）Action: 触发的事件，例如：切换客服状态、获取机器人问题等；11）Datares: ack机制，告诉网关已收到该消息。7.2、创建连接对网络层消息链接实例化，实现消息的正常发送和接收。其实现如下：7.3、消息定义客服要发送一条消息，肯定有对应的消息结构体model，即需要对消息体进行设计，这里会设计一下message类，每次创建新的消息体都会new一个实例，通过对实例的操作可以更新消息状态等。如下所示：针对单个消息，我们也要定义好消息状态，用于聊天过程中消息状态的更新。如下：7.4、数据池消息类创建好之后，就需要有消息数据池来存储。消息池结构定义如下：这里还涉及到消息体的一些基本操作方法对数据池中的数据进行操作，就不做过多的阐述。7.5、用户维度上面都是在分析公共模块，但是客服和用户是一对多的关系存在，还需要设计一个用户维度模块，后续在业务侧的操作基本都是以用户维度来操作，需要从单个用户维度设计对应的订阅关系、消息发送、删除等等。其实现大致如下：7.5.1发送消息链路分析针对客服发送消息，我们首先要站在客服角度考虑消息是否已发出去，优先展示的聊天页面，而不是等网关给了回复后在展示到聊天页面。根据已往经验来看，只要回车消息就要立即展示到聊天页面，否则客服会认为出现了卡顿，体验效果不佳，鉴于这种场景的需求，在设计发送消息链路的时候就要充分考虑到这一点。所以设计流程如下：如上图所示：先在SDK内进行处理对应的消息，处理完成后返回到业务侧完成渲染后再进行消息发送到网关，正常情况下都在一帧之内，客服是感知不到有延迟的。这里要关注消息体序列化、反序列化的时机，避免无谓的性能浪费。上述图中有个虚拟seq：主要是为了在未收到IM网关响应之前进行排序用的，比如图片、视频、断网发送消息、消息发送失败，或收到IM网关回复缺少seq（场景：敏感词）等情况都需要通过虚拟seq进行准确排序。7.5.2接收消息链路分析接收消息过程相对比较简单，收到消息进行反序列化后更新相关数据，然后在数据池中完成去重（重试机制）、排序后更新到业务侧渲染即可（如下图所示）。7.5.3消息的可靠传递IM消息的可靠投递主要是指：消息在发送接收过程中，能够做到不丢消息、消息不重复、消息顺序不错乱。我们先来分析以下2种情况。第一种情况：如果客服A在把消息发送到IM网关的过程中：1）由于网络不通等原因失败了；2）或者IM网关接收到消息进行存储时失败了；3）或者IM网关一直没有返回结果，导致超时。以上这些情况客服A都会被提示消息发送失败。第二种情况：消息在IM网关存储完后，客服A被告知消息发送成功了，然后IM网关把消息推送给用户A的在线设备：1）在推送的准备阶段或者把消息写入到内存后，如果服务端出现掉电，也会导致消息不能成功推送给用户A；2）如果用户A的设备在接收到消息，在后续处理过程中出现问题，也会导致消息丢失。针对第2）点，具体场景比如：用户A的设备在把消息写入本地DB时，出现异常导致落库失败，这种情况下，由于网络层面实际上已经成功传输，但用户A却看不到消息。我们客服IM对于消息丢失的处理方案主要是参考TCP协议的ACK机制，实现了一套基于业务层的ACK协议。添加ACK之前消息发送的时序图如下：7.5.3.1）ACK 机制：在TCP协议中，默认提供了ACK机制，通过一个协议自带的标准的ACK数据包，来对通信方接收的数据进行确认，告知通信发送方已确认成功接收了数据。ACK机制也是类似，需要解决的是：IM网关推送后如何确认消息是否成功送达接收方并明确被接收方所接收。具体实现的时序图如下：客服或用户在发送消息的过程中都会携带一个msgid（32位的uuid，类似TCP的sequenceId），IM网关在接收到消息后，会根据msgid到数据库中查询是否存在该条消息，如果存在就不落库，如果不存在就落库。然后再推送到接收方，接收方在收到消息后会回复ACK，ACK包中会携带上当前最新的seqid，IM网关收到ACK回复后会对最大的seqid进行更新。这里为什么要更新最大seqid呢？这么设计肯定有一定道理的，IM网关在收到发送方发送的消息后除了到数据库中检测该消息是否存在外，还会对比当前接收到消息的seq和最大seqid两者之间的差值，会把 [seq, seqid) 之间的数据全部推到接收方，正常情况下都是 [n, n-1)，如果IM网关没有收到接收方ACK，n-1就不会更新，推送的消息个数就大于1了。如果seq和seqid相等那就是发送方重复推送的消息，这个时候就不会向接收方推送。这里就涉及到了消息重试，继续向下分析吧。PS：有关IM消息ID或序列号生成的专题文章可以阅读：《IM消息ID技术专题(一)：微信的海量IM聊天消息序列号生成实践（算法原理篇）》《IM消息ID技术专题(三)：解密融云IM产品的聊天消息ID生成策略》《IM消息ID技术专题(四)：深度解密美团的分布式ID生成算法》《IM消息ID技术专题(五)：开源分布式ID生成器UidGenerator的技术实现》《IM消息ID技术专题(六)：深度解密滴滴的高性能ID生成器(Tinyid)》7.5.3.2）ACK机制中的消息重试：消息推给A的过程中丢失了怎么办，比如:1）A网络实际已经不可达，但IM网关还没有感知到（ping出现问题）；2）消息在中间网络途中被某些中间设备丢掉了。解决这个问题也是参考了TCP协议的重传机制。我们会在客服端、IM网关、用户端都维护一个超时计时器，一定时间内如果没有收到对方回的ACK包，会重新取出该消息进行重推。在重试一定次数后，如果还是没有收到ACK，视为放弃。前端代码结构和效果如下：上述图片中的数据只是模拟消息重试，真实场景中执行频次肯定要比这个时间更久一些。7.5.3.3）消息重复推送的问题：如果在一定时间内没有收到ACK包，就会触发重试机制。收不到ACK的情况有两种，除了推送的消息真正丢失导致A不回ACK外，还可能是A回的ACK包本身丢了。解决方案是：发送方在发送消息时携带一个msgid，msgid是全局唯一的，针对同一条重推的消息msgid不变，接收方根据这个唯一的msgid进行去重，这样经过去重后，对于A来说，在聊天界面是不会看到重复的消息，不影响使用体验。7.5.3.4）保证消息不会乱序：消息的一致性是非常重要的，在聊天过程中消息顺序不能错乱。我们是这样考虑的：1）以发送方的本地时间戳为序号，但是这样有比较大的问题，发送方的时间戳是可以被改动的，这种方式不可取；2）IM网关服务是集群部署，会通过topic和seqid做为唯一索引，在接收到消息落库之前会生成seqid，客服端和用户端接收到发送消息的回执时需要根据返回的seqid(IM网关自增)进行消息排序，这种方式可取。通过以上的分析：客服IM消息的可靠性就是通过ACK机制、重试机制、去重机制、排序机制来确保每一条消息的完整触达和准确排序。八、消息框架的分层实现：应用层业务侧使用的时候直接实例化SDK即可，在消息链路发布订阅中已经提到了RxJS，此时在业务侧订阅使用即可。需要注意的是在实例化SDK的时候传递了一个filterMsgItem方法，主要是为特殊业务场景提供使用的。就拿我们客服业务来说：有些特定消息是不需要展示到聊天页面的（比如：用户发送消息被篡改等）。当然我们在业务侧重新对数据过滤或者渲染的时候也是可以做过滤的，这样操作是没什么问题，但是没有必要，如果不从源头过滤数据，后续参与二分、倒序查找的源数据也会增加。会有一些不必要的浪费。当然也可以不添加这个参数，SDK都是全兼容的。至此我们就完成了整个SDK的实现以及在业务侧的使用，消息发送和接收也都正常。效果如下：九、本文小结自研IM SDK还是蛮有挑战的一件事情，从单纯的基于第三方SDK二次开发到自研SDK并与我们的实际业务场景相对完美的结合。在SDK的整体设计以及和业务侧如何更完美的结合并不是一蹴而就的，都是在实际业务场景中不断积累经验，不断尝试才找到相对完美的解决方案。这里列举一个简单的案例吧。例如消息发送，需要考虑到断网场景下：1）该如何进行消息显示、排序、重新发送？2）发送失败的场景下重新发送再次失败后又该如何显示、排序？3）弱网场景下发送消息触发重试机制该如何以最优的方式去重、排序？4）发送消息触发敏感词该如何处理？5）断网重连后对于发送失败和触发敏感词的消息又该如何处理？6）如果在涉及到文件又该如何处理？……在自研过程中除了关注业务场景外，还调研了行业内比较好的一些Web应用在某些特殊场景的处理方式。很多优秀的方案也都只能是借鉴一些核心思想，还是要以业务为核心，真正通过技术手段解决业务痛点才是最重要的。自研SDK收益还是非常大的，也积累了很多IM方面的经验，完成自研IM SDK也只是一个开始，后续我们将会在耗时任务、数据安全等方面持续深耕细作。十、参考资料[1] 从零到卓越：京东客服即时通讯系统的技术架构演进历程[2] 瓜子IM智能客服系统的数据架构设计（整理自现场演讲，有配套PPT）[3] 从游击队到正规军(一)：马蜂窝旅游网的IM系统架构演进之路[4] 一套高可用、易伸缩、高并发的IM群聊、单聊架构方案设计实践[5] 浅谈IM系统的架构设计[6] 简述移动端IM开发的那些坑：架构设计、通信协议和客户端[7] 一套海量在线用户的移动端IM架构设计实践分享(含详细图文)[8] 一套原创分布式即时通讯(IM)系统理论架构方案[9] 一套亿级用户的IM架构技术干货(上篇)：整体架构、服务拆分等[10] 从新手到专家：如何设计一套亿级消息量的分布式IM系统[11] 企业微信的IM架构设计揭秘：消息模型、万人群、已读回执、消息撤回等[12] 阿里IM技术分享(三)：闲鱼亿级IM消息系统的架构演进之路[13] 基于实践：一套百万消息量小规模IM系统技术要点总结[14] 跟着源码学IM(十)：基于Netty，搭建高性能IM集群（含技术思路+源码）[15] 一套十万级TPS的IM综合消息系统的架构实践与思考[16] 直播系统聊天技术(八)：vivo直播系统中IM消息模块的架构实践[17] 融云技术分享：全面揭秘亿级IM消息的可靠投递机制（本文已同步发布于：http://www.52im.net/thread-4153-1-1.html）

文章