0%

备注:本文完全为chatgpt生成。

快速幂算法(又称二分幂算法)是一种快速计算一个数的正整数次幂的算法,其时间复杂度为$O(logn)$,相较于朴素算法的时间复杂度$O(n)$,有很大的优势。下面是 Python 实现快速幂算法的示例代码:

1
2
3
4
5
6
7
8
9
10
def fast_power(x: int, n: int) -> int:
"""
使用快速幂算法计算x的n次方
"""
if n == 0:
return 1
elif n % 2 == 0:
return fast_power(x*x, n//2)
else:
return x*fast_power(x*x, (n-1)//2)

该函数的输入参数为 x 和 n,分别表示底数和指数。函数使用递归的方法来计算$x^n$,当指数为 0 时,返回 1;当指数为偶数时,将指数折半,递归计算$x^{n/2}$的平方;当指数为奇数时,先将指数减 1,然后递归计算$x^{(n-1)/2}$的平方,最后再乘以 x。这样就可以将$x^n$的计算分解成多个$x^{n/2}$的计算,从而实现了快速幂的效果。

下面是一个简单的示例,调用 fast_power 函数计算 2 的 10 次方:

1
2
result = fast_power(2, 10)
print(result) # 输出结果为:1024

可以看到,输出结果为 1024,即 2 的 10 次方的结果。

在上篇文章使用 opentelemetry 与 jaeger 实现 flask 应用的链路追踪 | 那时难决 (duyixian.cn)中,我们介绍了如何使用 opentelemetry 与 jaeger 对 flask 应用进行链路跟踪。

在真实的业务场景下多个服务间互相调用是十分常见的,在进行一些问题排查的时候有必要跟踪一个请求链路在各个服务中细节。

使用 opentelemetry 与 jaeger 同样可以实现跨应用的链路追踪。

原理

进行链路跟踪的核心概念是 trace,trace 是一个具有开始时间和结束时间的操作,它可以包含若干个 span,形成一个树状结构。
每一个 trace 都有一个唯一的 traceId,用于标识一个请求链路。在跨服务的场景下,我们可以通过 traceId 将一个请求链路中的所有 span 关联起来。

回到上一篇文章的场景,编辑main.py,定义/headers路由,使用 requests 库请求 https://httpbin.org/headers ,返回 requests 发起请求时的 header 信息。

1
2
3
@app.get("/headers")
def headers():
return requests.get("https://httpbin.org/headers").json()

headers

可以看到 headers 中有一个Traceparent,携带了 TraceId 信息。调用其他服务时,我们也需要将这个Traceparent传递给下游服务。

Read more »

链路追踪是应用可观测性的重要组成部分,它可以帮助我们快速定位问题,提高应用的可用性和稳定性。
今天我们来看看如何使用 opentelemetry 与 jaeger 实现 flask 应用的链路追踪。

OpenTelemetry 与 Jaeger

OpenTelemetry is a collection of tools, APIs, and SDKs. Use it to instrument, generate, collect, and export telemetry data (metrics, logs, and traces) to help you analyze your software’s performance and behavior.

OpenTelemetry 是一个用于收集、生成、导出遥测数据(metrics、logs、traces)的工具集合,它可以帮助我们分析软件的性能和行为。

Jaeger 是一个开源的分布式跟踪系统,它可以收集、存储和分析应用的链路追踪数据。

Jaeger 支持 Opentelemetry 协议,可以直接从 OpenTelemetry 收集数据。

在 Flask 中集成 OpenTelemetry 与 Jaeger

部署 Jaeger

Jaeger 有多种部署方式,在开发环境下最简单的方式是使用 Jaeger 官方提供的 all-in-one 镜像。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
## make sure to expose only the ports you use in your deployment scenario!
docker run -d --name jaeger \
-e COLLECTOR_OTLP_ENABLED=true \
-e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
-p 5775:5775/udp \
-p 6831:6831/udp \
-p 6832:6832/udp \
-p 5778:5778 \
-p 16686:16686 \
-p 14250:14250 \
-p 14268:14268 \
-p 14269:14269 \
-p 4317:4317 \
-p 4318:4318 \
-p 9411:9411 \
jaegertracing/all-in-one:1.42

访问 http://localhost:16686 查看 Jaeger 的 UI。

jaeger ui

Read more »

最近发现我三年前趁着活动使用优惠在腾讯云上购买的四核 8G 的云服务器在月底就要到期了。续费的价格十分昂贵,要三千多块钱一年。
仔细想想,我也没有什么需要在云服务器上运行的服务,基本上只有一个静态博客网站,然后就是偶尔在服务器上进行一些实验。
事实上这些需求完全可以借助 Serverless 平台来实现,而且大部分 Serverless 平台都有一定的免费额度,对于个人用途来说完全够用。

Vercel

Vercel 是一个 Serverless 平台,可以用来部署静态网站和 Serverless 函数。
Vercel 直接与 Github 集成,可以直接从 Github 仓库中部署网站,支持自定义域名,可以将自己的域名绑定到 Vercel 上(同时提供 TLS 证书),并且原生支持诸多静态网站生成器,例如 Hexo、Gatsby、Next.js 等。
我的博客(从云服务器上迁移过来)和知识库就是使用 Vercel 部署的。

blog on vercel

Github Actions

Github Actions 是 Github 提供的一个 CI/CD 服务,可以用来自动化构建、测试、部署应用。
我的几个 Python 项目(例如html_dsl)都配置了 Github Actions,可以在每次提交代码后自动运行持续集成,主要是运行静态代码分析和单元测试,以及在新版本打了 tag 后自动发布到 Pypi 上。
Github Action 也适用于运行一些定时任务。

github actions

Github Codespaces

Github Codespaces是 Github 提供的远程开发环境,可以通过 VS Code 或浏览器连接使用,并提供每个月 60 个小时的免费额度。

腾讯云函数计算 SCF

云函数 SCF是腾讯云的无服务器执行环境。我之前在腾讯云上部署过一些 Python 函数。
之前云函数还是有一定的免费额度的,但是现在已经取消了,每个月至少要购买 12.8 元的基础额度。
并且腾讯云的函数计算的 Python 版本支持不完善,只有 Python 3.7,远远落后于最新的 Python 3.11 版本。

腾讯云对象存储 COS

对象存储 COS是腾讯云的对象存储服务。我有利用腾讯云 COS 和 CDN 服务搭建一个图床,用于存储一些在博客中使用到的图片。

Deno Deploy

Deno Deploy出自 Deno 团队是一个 JavaScript,Typescript 和 WebAssembly 的运行时。我在 Deno Deploy 上部署了一个简单的接口general endpoint,可以用于进行一些简单的 http 接口测试。

general endpoint: sourcecode

general endpoint: api

Cloudflare Workers

Cloudflare Workers 是一个 Serverless 服务,可以在 Cloudflare 的全球 CDN 网络上运行 JavaScript 代码。

Cloudflare Workers KV

Cloudflare Workers KV 是一个 Key-Value 存储服务,可以在 Cloudflare Workers 中使用。
Workers KV 是一个十分难得的 Serverless 数据库服务,并且具有免费额度。毕竟云服务厂商的数据库服务都是十分昂贵的。


利用好各式 Serverless 服务,可以避免购买、维护服务器的工作,并且具有额外的优势,比如现成的 CDN 服务。

背景

目前有大量的项目是通过 Github Release 进行应用分发的,比如 Powershell Core 和一些新兴的命令行工具(bat,tokei 等)。

不过在国内的服务器中通过 Github Release 下载文件的速度和稳定性通常得不到保障。这时候我一般会选择在 PC 上开代理将文件下载到本地,然后通过 FileZilla 工具将文件上传到服务器上。

FileZilla(sftp)工具实际上是进行了 ssh 连接,要求在本地机器上拥有 ssh 私钥,在非个人设备上难以做到。

其实在服务器上部署一个文件上传服务可以有效解决这个需求。

于是我在前段时间在服务器上部署了一个 miniserve 服务,用于处理类似的文件上传下载需求。

miniserve

miniserve 是一个通过 http 服务托管文件和目录的工具,使用 Rust 实现,拥有很多开箱即用的特性(web 页面,上传文件,下载目录(以 zip 或 tar.gz 格式打包),用户名密码认证等)。

screenshot.png

部署 miniserve

启动 miniserve 服务

使用如下的命令启动 miniserve 服务。

1
miniserve -a $USER:$PASSWD -D -u -U -p $PORT -z -W --route-prefix=_files -t miniserve .

各个参数的含义分别为:

  • -a 设置访问用户名与密码
  • -D 将目录展示到文件之上
  • -u 允许上传文件
  • -U 允许创建子目录
  • -p 指定端口
  • -z 支持将整个目录通过 zip 下载
  • -W 在页面下方显示 wget 命令
  • –route-prefix 指定路径前缀
  • t 指定页面 title

启动服务后,可以通过localhost:$PORT访问 miniserve 的 web 页面。或者直接通过 wget 等工具直接进行文件上传或者下载。

配置 Nginx

不建议直接将 miniserve 服务直接通过 0.0.0.0 暴露到公网,毕竟上传或下载的文件总是会有一些私密性的,使用 https 是很有必要的。
miniserve 工具实际上是支持 tls 的,不过为一个单独的服务申请并配置证书总感觉有点浪费。
可以将 miniserve 服务通过配置了 https 的 nginx 进行端口转发。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
server {
server_name www.duyixian.cn;
rewrite ^/(.*)$ https://www.duyixian.cn:443/$1 permanent;
}

server{
listen 443;
server_name www.duyixian.cn wx.duyixian.cn;
ssl on;
ssl_certificate /cert/www.duyixian.cn_bundle.crt;
ssl_certificate_key /cert/www.duyixian.cn.key;
ssl_session_timeout 5m;
ssl_protocols TLSv1 TLSv1.1 TLSv1.2;
ssl_ciphers ECDHE-RSA-AES128-GCM-SHA256:HIGH:!aNULL:!MD5:!RC4:!DHE;
ssl_prefer_server_ciphers on;
error_page 404 /404.html;
location /miniserve/ {proxy_pass http://127.0.0.1:8080/miniserve/; }
location /dffd245af1 {proxy_pass http://127.0.0.1:8588; } # 处理css
}

一个小坑

miniserve 提供的网页引用了一个 css 文件,这个 css 实际上是内嵌在二进制文件里的,在每次运行的时候会随机生成一个名字。使用 nginx 转发 miniserve 服务的时候需要额外处理一下。

总结

miniserve 是一个相当实用的工具。除了作为文件服务器之外,还可以用来托管静态网站,甚至是单页应用(使用miniserve --spa --index index.html指定入口页面)。
也要感谢 rust 语言提供的开发效率提升和良好的生态,让广大的开发者可以使用 rust 改善传统的命令行工具。

背景

最近遇到了一个需要使用独占锁来保证业务正确性的场景,鉴于服务本身也会使用到 redis 缓存,可以直接利用 redis 提供的锁支持。

Redis Lock

基本使用

创建锁

1
2
3
4
from redis import Redis

client = Redis()
lock = client.lock(name="key",timeout=60.0,blocking_timeout=5.0)

可以直接调用 redis 实例的 lock 方法,并指定锁的名称,超时时间和等待时间(如果未能在 blocking_timeout 内获取到锁,会抛出异常)。

使用锁

1
2
3
lock.acquire()
do_something()
lock.release()

调用acquire方法获取锁,业务逻辑执行完成后调用release方法释放锁。

使用上下文管理器

手动获取并释放锁的使用方法比较繁琐,并且忘记调用acquire方法或因为业务逻辑异常导致acquire方法没有成功调用的风险。我们可以使用上下文管理器来更方便也更安全地使用 redis 锁。

1
2
with client.lock(name="key",timeout=60.0,blocking_timeout=5.0):
do_something()

这本身也是 RAII(资源获取即初始化) 思想的一个应用。

封装

我们可以使用functools.partial函数,对 redis 的 lock 使用进行一些简单的封装。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from functools import partial

def lock(
*,
prefix: str,
key: str,
timeout: float,
blocking: float = 5.0,
client: Redis = default_client,
) -> Lock:
return client.lock(f"{prefix}:{key}", timeout=timeout, blocking_timeout=blocking)


a_lock = partial(lock, prefix="A", timeout=60.0)

def do_a(uid: int) -> None:
with alock(key=uid):
pass

我们定义了一个lock函数,可以指定使用锁时的名称(由prefixkey组合而来)、超时时间、等待时间和使用的 redis 实例。
在具体的业务场景中,可以使用functools.partial函数定义更个性化的锁。
在这个示例中,我们定义了一个a_lock函数,指定了锁的名称前缀为A,超时时间为 60 秒。
在使用a_lock的时候,只需要指定key 即可。

总结

functools.partial实用性很高,值得在业务中使用。

【使用 Python 实现算法】目录


Python 标准库中的functoolsitertools模块,提供了一些函数式编程的工具函数。

functools 高阶函数

cache 与 lru_cache

用户缓存函数值的装饰器,可以缓存函数的调用结果。其中lru_cache函数可以设置一个缓存的最大容量,使用 LRU 算法淘汰长期不用的缓存。cache函数容量没有限制,相当于lru_cache(maxsize=None)

1
2
3
4
5
6
7
8
9
10
11
12
13
@lru_cache(maxsize=3)
def fib(n):
if n <= 1:
return n
return fib(n - 1) + fib(n - 2)

assert [fib(n) for n in range(10)] == [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

@cache
def factorial(n):
return 1 if n == 0 else n * factorial(n - 1)

assert [factorial(n) for n in range(10)] == [1, 1, 2, 6, 24, 120, 720, 5040, 40320, 362880]

partial 固定函数的一部分参数

partial() 会被“冻结了”一部分函数参数和/或关键字的部分函数应用所使用,从而得到一个具有简化签名的新对象。 例如,partial() 可用来创建一个行为类似于 int() 函数的可调用对象,其中 base 参数默认为二:

1
2
3
4
from functools import partial
basetwo = partial(int, base=2)
basetwo.__doc__ = 'Convert base 2 string to an int.'
assert basetwo('10010') == 18

reduce

Python2 中的reduce函数是一个内置函数,在 Python3 中被移入了functools模块中。
原因是reduce函数的效率没有寻常的 for 循环高,而往往使用了reduce函数的代码也不是特别容易理解(对比mapfilter)。

1
assert reduce(lambda x, y: x + y, [1, 2, 3, 4, 5]) == 15

itertools 实用的迭代器

itertools模块是 Python 参考 APL,Haskell,SML 等编程语言的启发,实现的个快速、高效利用内存的核心工具集。
在算法实现中利用好itertools模块,可以有效提高代码编写速度、可维护性和算法的效率。

islice

首先介绍一下islice函数,可用于迭代器切片。

1
assert list(islice('ABCDEFG', 3)) == ['A', 'B', 'C']

无穷迭代器

count

count函数可以生成一个无穷的迭代器,每次调用返回一个整数,从 0 开始,每次增 1。
可以指定初始值,步长和生成数量。
count函数很适合替代while True:n +=1的循环。

1
2
3
4
5
6
7
8
def gen_primes():
primes = []
for n in count(2):
if not any(n % p == 0 for p in primes):
primes.append(n)
yield n

assert list(islice(gen_primes(), 10)) == [2, 3, 5, 7, 11, 13, 17, 19, 23, 29]

cycle

cycle函数可以不断地迭代一个序列,直到迭代完成,然后重新开始迭代。

1
assert list(islice(cycle("ABC"), 6)) == ["A", "B", "C", "A", "B", "C"]

repeat

repeat函数会不断地重复生成一个元素。

1
assert list(map(pow, range(6), repeat(3))) == [0, 1, 8, 27, 64, 125]

有限数量的迭代器

accumulate

accumulate可以简单地理解为一个累加器。在很多需要计算列表的前缀和的算法中很有用。

1
assert list(islice(accumulate(count()), 5)) == [0, 1, 3, 6, 10]

chain

chain函数可以理解为合并多个序列的迭代器。

1
st(chain("ABC", "DEF")) == ["A", "B", "C", "D", "E", "F"]

dropwhile

dropwhile可以删除序列中的前缀元素,直到某个条件不满足。

1
assert list(dropwhile(lambda x: x < 5, [1, 4, 6, 4, 1])) == [6, 4, 1]

filterfalse

filterfalse可以过滤出一个序列中不满足某个条件的元素。

1
assert list(filterfalse(lambda x: x % 2 == 0, range(10))) == [1, 3, 5, 7, 9]

pairwise

pairwise类似于一个容量为 2 的滑动窗口,每次迭代返回一对元素。

1
asset list(pairwise([1, 2, 3, 4, 5])) == [(1, 2), (2, 3), (3, 4), (4, 5)]

takewhile

takewhile遍历一个序列,直到第一个不满足某个条件的元素。

1
assert list(takewhile(lambda x: x < 5, [1, 4, 6, 4, 1])) == [1, 4]

排列组合迭代器

product

product可以得到多个序列的笛卡尔积。

1
assert list(product("ABCD", "xy")) == [('A', 'x'), ('A', 'y'), ('B', 'x'), ('B', 'y'), ('C', 'x'), ('C', 'y'), ('D', 'x'), ('D', 'y')]

permutations

permutations可以得到一个序列的全排列。

1
2
assert list(permutations([1, 2, 3])) == [(1, 2, 3), (1, 3, 2), (2, 1, 3), (2, 3, 1), (3, 1, 2), (3, 2, 1)]
assert list(permutations([1, 2, 3], 2)) == [(1, 2), (1, 3), (2, 1), (2, 3), (3, 1), (3, 2)]

combinations

combinations可以得到一个序列的组合。

1
assert list(combinations(range(4),2)) == [(0, 1), (0, 2), (0, 3), (1, 2), (1, 3), (2, 3)]

combinations_with_replacement

combinations_with_replacement可以得到一个序列的可重复组合。

1
list(combinations_with_replacement([1,2,3],2)) == [(1, 1), (1, 2), (1, 3), (2, 2), (2, 3), (3, 3)]

总结

使用 Python 实现算法相比其他语言的一个优势就是标准库中的itertools,可以节省大量编写 for 循环和递归函数的时间。

【使用 Python 实现算法】目录


算法实现中经常需要构造和处理一些特殊的数据结构,Python 标准库中有一些模块可以帮到我们。

collections 实用容器

collections模块实现了一些特定目标的容器,以提供 Python 标准内建容器 dict , list , set , 和 tuple 的替代选择。

deque

在 Python 中从list对象的头部删除元素,时间复杂度是 O(n)。deque类型是一个双向队列,类似列表(list)的容器,实现了在两端快速添加(append)和弹出(pop)。

1
2
3
4
5
6
7
8
9
q = deque([1, 2, 3, 4])

q.append(5)
q.appendleft(0)
assert q == deque([0, 1, 2, 3, 4, 5])

assert q.pop() == 5
assert q.popleft() == 0
assert len(q) == 4
Read more »

【使用 Python 实现算法】目录


本期向大家介绍一些 Python 中用于处理数字和数学方面的标准库。

math 数学模块

作为 LeetCode Python 环境中默认导入的标准库模块之一,math模块提供了很多非常有用的数字和数学方面的函数。

数论与表示函数(number theoretic and representation functions)

gcdlcm,用于计算多个整数的最大公约数与最小公倍数。

1
2
3
4
5
assert math.gcd(12, 18) == 6
assert math.gcd(3, 5, 7) == 1

assert math.lcm(12, 18) == 36
assert math.lcm(3, 5, 7) == 105

ceil:向上取整,floor:向下取整,trunc:移除小数部分。
对于正数来说,truncfloor是等价的。
对于负数来说,truncceil是等价的。

1
2
3
4
5
6
7
assert math.ceil(1.2) == 2
assert math.floor(1.2) == 1
assert math.trunc(1.2) == 1

assert math.ceil(-3.5) == -3
assert math.floor(-3.5) == -4
assert math.trunc(-3.5) == -3
Read more »

【使用 Python 实现算法】目录


本期话题是 Python 的原生类型和内置函数在算法实现中的一些技巧,首先从最常见的 Python 原生类型开始。

int

int类型有两个内置的方法bit_countbit_length,分别用来获取整数二进制表达的1的个数和二进制位数,在很多场景下都很实用。

1
2
3
n = 9
assert n.bit_count() == 2
assert n.bit_length() == 4

float

floatas_integer_ratio方法可以获得浮点数的最简分数表达。

1
2
f = 6.125
assert f.as_integer_ratio() == (49, 8)

利用f-string可以轻松获得浮点数的指定小数位数的字符串表达。

1
assert f"{1/3:.4f}" == "0.3333"
Read more »