博观网

Ubuntu中设置网络IP地址4种方法

Tue, 18 Nov 2025 17:41:18 +0800

在Ubuntu中设置网络IP地址有多种方法，详细介绍几种常见的方式：

方法1：使用netplan（Ubuntu 17.10及以后版本）

设置静态IP

查看网络接口名称

ip addr show
# 或
nmcli device status

编辑netplan配置文件

sudo nano /etc/netplan/01-netcfg.yaml

配置静态IP示例

network:
  version: 2
  renderer: networkd
  ethernets:
    ens33:
      addresses:
        - 192.168.1.100/24
      routes:
        - to: default
          via: 192.168.1.1
      nameservers:
        addresses: [8.8.8.8, 1.1.1.1]

应用配置

sudo netplan apply

设置为DHCP

network:
  version: 2
  renderer: networkd
  ethernets:
    ens33:
      dhcp4: true
      dhcp6: true

方法2：使用NetworkManager（桌面版常用）

图形界面设置

打开网络设置

nm-connection-editor
# 或点击系统托盘的网络图标

选择对应连接 → 编辑 → IPv4设置
- 手动：设置静态IP
- 自动(DHCP)：使用DHCP

命令行设置

设置静态IP

sudo nmcli con mod "有线连接 1" ipv4.addresses "192.168.1.100/24"
sudo nmcli con mod "有线连接 1" ipv4.gateway "192.168.1.1"
sudo nmcli con mod "有线连接 1" ipv4.dns "8.8.8.8,1.1.1.1"
sudo nmcli con mod "有线连接 1" ipv4.method manual
sudo nmcli con down "有线连接 1"
sudo nmcli con up "有线连接 1"

设置为DHCP

sudo nmcli con mod "有线连接 1" ipv4.method auto
sudo nmcli con down "有线连接 1"
sudo nmcli con up "有线连接 1"

方法3：传统ifupdown方法（较老版本）

设置静态IP

编辑接口配置

sudo nano /etc/network/interfaces

配置内容

auto ens33
iface ens33 inet static
address 192.168.1.100
netmask 255.255.255.0
gateway 192.168.1.1
dns-nameservers 8.8.8.8 1.1.1.1

设置为DHCP

auto ens33
iface ens33 inet dhcp

重启网络服务

sudo systemctl restart networking

方法4：使用ip命令（临时设置）

临时设置静态IP

sudo ip addr add 192.168.1.100/24 dev ens33
sudo ip route add default via 192.168.1.1

临时设置为DHCP

sudo dhclient ens33

验证配置

设置完成后，使用以下命令验证：

# 查看IP配置
ip addr show

# 测试网络连通性
ping -c 4 8.8.8.8

# 查看路由表
ip route show

# 测试DNS解析
nslookup google.com

注意事项

备份配置文件：修改前建议备份原配置
接口名称：根据实际接口名称修改（ens33、eth0等）
网络参数：根据实际网络环境修改IP、网关等参数
权限：需要sudo权限修改系统网络配置

推荐使用netplan方法，这是Ubuntu当前主推的网络配置工具。

Python代码加密的四种方案

Mon, 20 Oct 2025 17:15:22 +0800

在交付 Python 项目时，我们通常不希望源代码被轻易查看、修改或窃取。

首先，需要明确一个核心概念：Python 作为一种解释型语言，不存在像 C++/Java 那样编译成机器码后能实现“绝对”加密的方法。 我们所有的努力，都是为了增加逆向工程的难度和成本，而不是做到完全无法破解。

下面我将从易到难，从“混淆”到“编译”，介绍几种免费的方式，并分析它们的优缺点。

方案一：代码混淆

这是最简单、最直接的方式，不改变代码的执行逻辑，但让代码变得极难阅读和理解。

1. 手动混淆

通过一些简单的约定和技巧，增加代码的阅读障碍。

变量/函数名混淆：将有意义的名称（如 calculate_user_balance）替换为无意义的短名称（如 x_y_z）。
移除注释和文档字符串：删除所有对代码逻辑的解释。
使用__all__ ：限制 from module import * 时导入的符号，但这更多是规范而非加密。
字符串加密：将关键的字符串（如密码、URL）进行编码或加密，在运行时再解码。

优点：

完全免费，无需任何第三方库。
简单快捷，可以立即实施。

缺点：

安全性极低：对于有经验的程序员来说，通过单步调试、打印变量等方式，很容易就能理清逻辑。
破坏了代码的可维护性，你自己后续维护也会很痛苦。

适用场景：只想防止“小白”用户直接查看代码，对安全性要求不高的场景。

2. 自动混淆工具

使用工具自动完成上述混淆过程，效果更好，也更省力。

推荐工具：pyminifier

这是一个非常流行的 Python 代码混淆工具。

安装：

pip install pyminifier

使用示例：

假设你有一个 my_script.py 文件。

# 基本混淆：移除注释，压缩空格

pyminifier --obfuscate my_script.py > my_script_obfuscated.py

# 更强混淆：替换变量名

pyminifier --obfuscate-variables my_script.py > my_script_obfuscated_vars.py

# 最强混淆：替换变量名，并使用 Base64 编码字符串

pyminifier --obfuscate-variables --obfuscate-classes --obfuscate-functions --replacement-length=1 my_script.py > my_script_obfuscated_max.py

优点：

免费、开源。
比手动混淆更彻底、更高效。
可以集成到自动化构建流程中。

缺点：

依然不是真正的加密。有经验的攻击者可以通过 ast (抽象语法树) 分析或动态调试来还原逻辑。
可能会引入一些意想不到的 bug，特别是当代码中大量使用 eval()、exec() 或动态属性访问时。

适用场景：希望提高代码阅读门槛，防止代码被轻易复制和修改的场景。

方案二：编译成字节码

Python 在执行 .py 文件时，会先将其编译成 .pyc 字节码文件（存储在 __pycache__ 目录中）。我们可以直接分发 .pyc 文件，而不是 .py 源文件。

1. 手动编译并分发 `.pyc`

操作步骤：

编译你所有的 Python 文件。
只分发 .pyc 文件，删除 .py 文件。

如何编译：

import py_compile

# 编译单个文件

py_compile.compile('my_module.py', cfile='my_module.pyc', optimize=2)

# 或者，使用 compileall 编译整个目录

import compileall

# optimize=2 表示移除断言和文档字符串

compileall.compile_dir('.', force=True, optimize=2)

优点：

Python 内置功能，完全免费。
隐藏了源代码的文本内容。
略微提升启动速度（省去了编译步骤）。

缺点：

非常容易被反编译。有现成的工具（如 uncompyle6）可以轻松地将 .pyc 文件反编译回几乎与原始代码一模一样的 .py 文件。
```
pip install uncompyle6

uncompyle6 my_module.pyc > my_module_decompiled.py
```
跨版本兼容性问题。用 Python 3.10 编译的 .pyc 文件无法在 Python 3.11 上运行。

适用场景：防止用户“无意中”看到源代码，或者对启动速度有微小要求的场景。不推荐作为主要的加密手段。

方案三：打包成可执行文件

这是目前最流行、最有效的免费交付方式。它将你的 Python 代码、依赖库、甚至一个 Python 解释器一起打包成一个独立的可执行文件（如 Windows 的 .exe，Linux 的可执行文件）。

推荐工具：PyInstaller

PyInstaller 是功能最强大、社区最活跃的打包工具之一。

安装：

pip install pyinstaller

使用示例：

# 将 my_script.py 打包成一个单文件可执行程序

pyinstaller --onefile --noconsole my_script.py

# --onefile: 生成一个独立的 .exe 文件

# --noconsole (或 -w): 运行时不显示黑色的控制台窗口（适用于GUI程序）

# --add-data: 添加非代码文件（如图片、配置文件）

# --icon: 指定程序图标

优点：

用户体验极佳：用户无需安装 Python 和任何依赖，直接运行即可。
安全性较高：代码被打包在可执行文件中，虽然仍可被逆向，但难度远高于 .pyc 反编译。攻击者需要使用专业的反汇编工具（如 Ghidra, IDA Pro）来分析，成本大大增加。
完全免费、开源。
跨平台支持。

缺点：

文件体积大：因为内置了解释器和依赖，单文件模式可能会产生几十MB甚至上百MB的文件。
杀毒软件误报：某些打包方式可能会被杀毒软件误判为病毒，需要进行代码签名或提交白名单。
启动速度可能比直接运行 .py 慢，因为需要解压资源到临时目录。
并非坚不可摧：专业的逆向工程师仍然可以从中提取出 .pyc 文件并进行反编译。

适用场景：绝大多数商业软件交付和内部工具分发的首选方案。 在安全性和易用性之间取得了最好的平衡。

方案四：编译成原生代码

这是最高级别的保护方式，它将 Python 代码真正地编译成特定平台的机器码。

推荐工具：Nuitka

Nuitka 是一个 Python 编译器，它将 Python 代码和模块转换成 C 代码，然后使用标准的 C/C++ 编译器将其编译成可执行文件或动态链接库。

安装：

pip install nuitka

使用示例：

# 编译成可执行文件

python -m nuitka --standalone --onefile --follow-imports my_script.py

# --standalone: 创建独立的可执行文件

# --onefile: 打包成单个文件

# --follow-imports: 递归编译所有导入的模块

优点：

目前免费方案中最高的安全性。生成的是原生机器码，反编译难度极大，接近于破解一个 C++ 程序。
性能提升：由于是编译成机器码，通常比标准的 CPython 解释执行有显著的性能提升。
兼容性好，生成的程序与 PyInstaller 类似，可以独立运行。

缺点：

编译过程复杂且耗时：需要系统上安装 C++ 编译器（如 Windows 上的 MSVC，Linux 上的 GCC）。
对某些库的兼容性可能不如 PyInstaller，特别是那些使用了 C 扩展或复杂动态导入机制的库。
仍然免费，但学习曲线比 PyInstaller 陡峭。

适用场景：对性能有较高要求，且对代码安全性有极致追求的商业软件。

总结与对比

方案	免费工具	安全性	易用性	性能影响	推荐指数
代码混淆	`pyminifier`	低	高	几乎无	★★☆☆☆
字节码编译	`py_compile`	极低	高	略微提升	★☆☆☆☆
打包可执行文件	`PyInstaller`	中高	高	启动稍慢	★★★★★
编译原生代码	`Nuitka`	高	中	性能提升	★★★★☆

最终建议

对于绝大多数 Python 项目交付场景：

**首选 PyInstaller**。它在安全性、易用性和用户体验之间取得了最佳平衡。对于 99% 的用户和客户来说，这已经足够安全了。
如果你的项目对性能有极致要求，或者你面对的是技术能力很强的对手，希望进一步提高破解门槛，那么**投入时间学习并使用 Nuitka**。
pyminifier 可以作为 PyInstaller 或 Nuitka 打包前的预处理步骤，实现“混淆+编译”的双重保护，让代码更难被分析。

重要提醒：无论你选择哪种方式，都无法做到 100% 防破解。软件保护的核心是提高破解成本，使其超过破解者能获得的利益。同时，配合软件许可验证、在线激活等业务逻辑层面的保护，才能构成更完整的安全体系。

Docker 与 Docker Compose 安装指南：使用代理加速下载

Thu, 18 Sep 2025 11:35:28 +0800

Docker 与 Docker Compose 安装指南：使用代理加速下载

Docker 已经成为现代应用开发和部署的标准工具之一。然而，在国内网络环境下，直接从官方源下载 Docker 及相关镜像可能会遇到速度慢甚至无法连接的问题。本文将详细介绍如何安装 Docker 和 Docker Compose，并特别说明如何配置代理服务来加速下载过程。

第一部分：Docker 安装

标准安装步骤

更新系统包索引
```
sudo apt-get update
```

安装依赖包

sudo apt-get install \
   apt-transport-https \
   ca-certificates \
   curl \
   gnupg \
   lsb-release

添加 Docker 官方 GPG 密钥

curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg

设置稳定版仓库

echo \
  "deb [arch=amd64 signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/ubuntu \
  $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null

安装 Docker 引擎

sudo apt-get update
sudo apt-get install docker-ce docker-ce-cli containerd.io

使用代理加速下载

如果在上述步骤中遇到下载缓慢的问题，可以通过以下方法配置代理：

方法一：临时使用代理下载

在执行 apt-get 命令前设置环境变量：

sudo http_proxy=http://your-proxy-address:port https_proxy=http://your-proxy-address:port apt-get update
sudo http_proxy=http://your-proxy-address:port https_proxy=http://your-proxy-address:port apt-get install docker-ce docker-ce-cli containerd.io

方法二：配置系统级代理

编辑 /etc/apt/apt.conf.d/proxy.conf 文件（如果不存在则创建）：

sudo nano /etc/apt/apt.conf.d/proxy.conf

添加以下内容：

Acquire {
  HTTP::proxy "http://your-proxy-address:port";
  HTTPS::proxy "http://your-proxy-address:port";
}

方法三：使用国内镜像源

如果不想使用代理，可以替换 Docker 下载源为国内镜像：

编辑 Docker 源列表文件：

sudo nano /etc/apt/sources.list.d/docker.list

将网址替换为国内镜像，例如：

deb [arch=amd64] https://mirrors.aliyun.com/docker-ce/linux/ubuntu/ $(lsb_release -cs) stable

验证安装

安装完成后，运行以下命令验证 Docker 是否安装成功：

sudo docker run hello-world

如果看到欢迎信息，说明 Docker 已正确安装。

第二部分：Docker Compose 安装

标准安装步骤

下载 Docker Compose 二进制文件

sudo curl -L "https://github.com/docker/compose/releases/download/v2.27.1/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose

应用可执行权限

sudo chmod +x /usr/local/bin/docker-compose

验证安装
```
docker-compose --version
```

使用代理加速下载

对于 Docker Compose 的下载，可以通过以下方式使用代理：

方法一：使用 curl 代理参数

sudo curl -x http://your-proxy-address:port -L "https://github.com/docker/compose/releases/download/v2.27.1/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose

方法二：先下载到本地再上传

如果代理方法仍然缓慢，可以：

在浏览器中手动下载 Docker Compose 二进制文件
通过 SCP 或 SFTP 上传到服务器

移动到指定位置并设置权限：

sudo mv docker-compose /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose

第三部分：配置 Docker 使用代理拉取镜像

即使安装了 Docker，拉取镜像时可能仍然需要代理。可以通过以下方式配置：

创建 Docker 服务代理配置目录

sudo mkdir -p /etc/systemd/system/docker.service.d

创建代理配置文件

sudo nano /etc/systemd/system/docker.service.d/http-proxy.conf

添加代理配置

[Service]
Environment="HTTP_PROXY=http://your-proxy-address:port"
Environment="HTTPS_PROXY=http://your-proxy-address:port"
Environment="NO_PROXY=localhost,127.0.0.1,.example.com"

重新加载配置并重启 Docker

sudo systemctl daemon-reload
sudo systemctl restart docker

验证代理是否生效

sudo systemctl show --property=Environment docker

第四部分：常见问题解决

代理认证问题
如果代理需要认证，在配置中使用以下格式：
```
http://username:password@proxy-host:proxy-port
```
权限问题
如果遇到权限错误，将当前用户添加到 docker 组：
```
sudo usermod -aG docker $USER
```
然后退出重新登录使更改生效。或者
```
su - $USER
```

Python asyncio 用法深度解析

Tue, 02 Sep 2025 15:12:53 +0800

引言

在现代软件开发中，处理高并发I/O操作是一个常见挑战。传统的同步编程模型在面对大量网络请求、文件操作或数据库查询时往往效率低下。Python的asyncio库应运而生，它提供了基于async/await语法的异步编程能力，让开发者能够编写高效、可扩展的并发代码。

本文将深入探讨asyncio的核心概念、用法、最佳实践以及常见陷阱，帮助你全面掌握这一强大的并发编程工具。

一、核心概念解析

1.1 事件循环 (Event Loop)

事件循环是asyncio的心脏，它负责调度和执行异步任务。想象事件循环就像一个高效的交通指挥员，它不断检查哪些协程可以运行，并在它们之间智能切换。

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# 现代Python推荐方式（3.7+）
asyncio.run(main())

# 传统方式（了解即可）
loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

1.2 协程 (Coroutine)

协程是asyncio的基本执行单元，使用async def定义。它们与普通函数的区别在于能够被”暂停”和”恢复”。

async def fetch_data():
    print("开始获取数据")
    await asyncio.sleep(2)  # 模拟I/O操作
    print("数据获取完成")
    return {"data": "sample"}

# 调用协程必须使用await
async def main():
    result = await fetch_data()
    print(f"结果: {result}")

1.3 Future 和 Task

Future: 表示异步操作的最终结果，类似于JavaScript中的Promise
Task: Future的子类，专门用于包装和管理协程

async def my_task():
    await asyncio.sleep(1)
    return "任务完成"

async def main():
    # 创建任务
    task = asyncio.create_task(my_task())
    
    # 可以继续执行其他操作
    print("任务已创建，继续其他工作...")
    
    # 等待任务完成
    result = await task
    print(f"任务结果: {result}")

二、基本用法详解

2.1 并发执行多个任务

asyncio的真正威力在于能够轻松实现并发操作：

import asyncio
import time

async def download_file(filename, duration):
    print(f"开始下载 {filename}")
    await asyncio.sleep(duration)
    print(f"{filename} 下载完成")
    return f"{filename}_content"

async def main():
    start_time = time.time()
    
    # 方法1: 使用gather（推荐）
    results = await asyncio.gather(
        download_file("file1.txt", 2),
        download_file("file2.txt", 3),
        download_file("file3.txt", 1)
    )
    print(f"所有文件下载完成: {results}")
    
    # 方法2: 分别创建任务
    task1 = asyncio.create_task(download_file("file4.txt", 2))
    task2 = asyncio.create_task(download_file("file5.txt", 1))
    
    await task1
    await task2
    
    end_time = time.time()
    print(f"总耗时: {end_time - start_time:.2f}秒")

asyncio.run(main())

2.2 超时处理

在实际应用中，超时控制至关重要：

async def slow_operation():
    await asyncio.sleep(10)
    return "操作完成"

async def main():
    try:
        # 设置5秒超时
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
        print(result)
    except asyncio.TimeoutError:
        print("操作超时，已取消")
    
    # Python 3.11+ 的新语法
    try:
        async with asyncio.timeout(5.0):
            result = await slow_operation()
            print(result)
    except TimeoutError:
        print("操作超时")

asyncio.run(main())

2.3 同步原语

asyncio提供了线程类似的同步机制：

async def worker(lock, name, work_time):
    async with lock:  # 获取锁
        print(f"{name} 开始工作")
        await asyncio.sleep(work_time)
        print(f"{name} 工作完成")

async def main():
    lock = asyncio.Lock()
    
    # 三个 worker 竞争同一个锁
    await asyncio.gather(
        worker(lock, "Worker1", 2),
        worker(lock, "Worker2", 1),
        worker(lock, "Worker3", 3)
    )

asyncio.run(main())

三、高级用法

3.1 队列 (Queue)

生产者-消费者模式是并发编程的常见模式：

async def producer(queue, items):
    for item in items:
        await queue.put(item)
        print(f"生产了: {item}")
        await asyncio.sleep(0.1)  # 模拟生产时间
    await queue.put(None)  # 结束信号

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            queue.put(None)  # 传递给其他消费者
            print(f"{name} 结束工作")
            break
        
        print(f"{name} 消费了: {item}")
        await asyncio.sleep(0.2)  # 模拟处理时间
        queue.task_done()  # 标记任务完成

async def main():
    queue = asyncio.Queue(maxsize=3)  # 限制队列大小
    
    # 启动生产者和消费者
    producer_task = asyncio.create_task(producer(queue, range(5)))
    consumer_tasks = [
        asyncio.create_task(consumer(queue, "消费者1")),
        asyncio.create_task(consumer(queue, "消费者2"))
    ]
    
    await producer_task
    await queue.join()  # 等待所有任务处理完成
    
    for task in consumer_tasks:
        task.cancel()
    
    # 等待消费者任务正常结束
    await asyncio.gather(*consumer_tasks, return_exceptions=True)

asyncio.run(main())

3.2 事件 (Event) 和条件 (Condition)

async def waiter(event, name):
    print(f"{name} 等待事件触发")
    await event.wait()
    print(f"{name} 检测到事件，继续执行")

async def setter(event):
    await asyncio.sleep(2)
    print("设置事件")
    event.set()  # 触发所有等待的协程

async def main():
    event = asyncio.Event()
    
    await asyncio.gather(
        waiter(event, "任务1"),
        waiter(event, "任务2"),
        setter(event)
    )

asyncio.run(main())

3.3 子进程管理

async def run_command():
    # 创建子进程
    process = await asyncio.create_subprocess_exec(
        'python', '--version',
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    
    # 等待进程完成并获取输出
    stdout, stderr = await process.communicate()
    
    print(f"退出码: {process.returncode}")
    if stdout:
        print(f"标准输出: {stdout.decode()}")
    if stderr:
        print(f"错误输出: {stderr.decode()}")

asyncio.run(run_command())

四、实际应用示例

4.1 并发HTTP请求

import aiohttp
import asyncio

async def fetch_url(session, url):
    try:
        async with session.get(url, timeout=10) as response:
            return await response.text()
    except Exception as e:
        return f"错误: {e}"

async def main():
    urls = [
        "http://httpbin.org/get",
        "http://httpbin.org/delay/2",
        "http://httpbin.org/status/404"
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for url, result in zip(urls, results):
            print(f"{url}: {result[:100]}...")

asyncio.run(main())

4.2 WebSocket客户端

import aiohttp
import asyncio

async def websocket_client():
    try:
        async with aiohttp.ClientSession() as session:
            async with session.ws_connect("ws://echo.websocket.org") as ws:
                print("连接已建立")
                
                # 发送消息
                await ws.send_str("Hello, WebSocket!")
                
                # 接收消息
                async for msg in ws:
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        print(f"收到消息: {msg.data}")
                        if msg.data == "close":
                            await ws.close()
                            break
                    elif msg.type == aiohttp.WSMsgType.CLOSED:
                        print("连接关闭")
                        break
                    elif msg.type == aiohttp.WSMsgType.ERROR:
                        print("连接错误")
                        break
    except Exception as e:
        print(f"连接失败: {e}")

asyncio.run(websocket_client())

五、注意事项与最佳实践

5.1 避免阻塞事件循环

这是asyncio开发中最常见的错误：

# ❌ 错误示例 - 阻塞调用
async def bad_example():
    import time
    time.sleep(5)  # 这会阻塞整个事件循环！

# ✅ 正确做法 - 使用异步版本
async def good_example():
    await asyncio.sleep(5)  # 非阻塞

# ✅ 对于无法异步的阻塞操作
async def run_blocking():
    loop = asyncio.get_running_loop()
    # 在线程池中运行阻塞函数
    result = await loop.run_in_executor(None, time.sleep, 5)
    return result

5.2 正确的错误处理

async def risky_operation():
    try:
        await asyncio.sleep(1)
        raise ValueError("模拟错误")
    except ValueError as e:
        print(f"操作失败: {e}")
        raise  # 重新抛出

async def main():
    # 单个任务的错误处理
    try:
        await risky_operation()
    except ValueError as e:
        print(f"捕获错误: {e}")
    
    # 多个任务的错误处理
    tasks = [
        asyncio.create_task(risky_operation()),
        asyncio.create_task(asyncio.sleep(2))
    ]
    
    done, pending = await asyncio.wait(tasks, return_exceptions=True)
    
    for task in done:
        if isinstance(task, Exception):
            print(f"任务失败: {task}")
        else:
            print(f"任务成功: {task.result()}")

asyncio.run(main())

5.3 资源清理

async def fetch_with_resource():
    # 模拟资源获取
    resource = "connection"
    print(f"获取资源: {resource}")
    
    try:
        # 模拟业务操作
        await asyncio.sleep(1)
        return "success"
    finally:
        # 确保资源释放
        print(f"释放资源: {resource}")

async def main():
    result = await fetch_with_resource()
    print(f"操作结果: {result}")

asyncio.run(main())

5.4 调试技巧

async def debug_example():
    # 获取当前任务信息
    task = asyncio.current_task()
    print(f"任务名称: {task.get_name()}")
    print(f"协程: {task.get_coro()}")
    
    await asyncio.sleep(1)

async def main():
    # 启用调试模式
    asyncio.run(debug_example(), debug=True)
    
    # 或者设置事件循环的调试模式
    loop = asyncio.get_event_loop()
    loop.set_debug(True)
    
    try:
        await debug_example()
    finally:
        loop.set_debug(False)

asyncio.run(main())

六、性能优化建议

6.1 限制并发数

async def limited_concurrency(urls, max_concurrent=5):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def fetch_with_limit(url):
        async with semaphore:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    return await response.text()
    
    tasks = [fetch_with_limit(url) for url in urls]
    return await asyncio.gather(*tasks)

# 使用示例
async def main():
    urls = ["http://example.com"] * 100
    results = await limited_concurrency(urls, max_concurrent=10)
    print(f"完成了 {len(results)} 个请求")

asyncio.run(main())

6.2 使用连接池

async def efficient_requests():
    connector = aiohttp.TCPConnector(limit=10)  # 限制连接数
    
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [session.get(f"http://example.com/page/{i}") for i in range(100)]
        responses = await asyncio.gather(*tasks)
        
        # 处理响应
        results = []
        for response in responses:
            results.append(await response.text())
        
        return results

6.3 批量操作

async def batch_operations():
    # 而不是这样：
    # results = []
    # for item in items:
    #     result = await process_item(item)
    #     results.append(result)
    
    # 应该这样：
    tasks = [process_item(item) for item in items]
    results = await asyncio.gather(*tasks)
    return results

七、常见问题解答

Q1: 如何从同步代码调用异步函数？

def sync_wrapper():
    # 方法1: 使用asyncio.run（适用于简单脚本）
    result = asyncio.run(async_function())
    return result

# 方法2: 在已有事件循环中运行
def sync_in_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        result = loop.run_until_complete(async_function())
        return result
    finally:
        loop.close()

Q2: 如何正确取消任务？

async def cancellable_task():
    try:
        await asyncio.sleep(10)
        print("任务正常完成")
        return "success"
    except asyncio.CancelledError:
        print("任务被取消，执行清理操作")
        raise  # 重新抛出异常

async def main():
    task = asyncio.create_task(cancellable_task())
    
    # 等待一段时间后取消
    await asyncio.sleep(2)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("任务已成功取消")

asyncio.run(main())

Q3: 如何处理任务中的异常？

async def failing_task():
    await asyncio.sleep(1)
    raise ValueError("任务失败")

async def main():
    task = asyncio.create_task(failing_task())
    
    try:
        await task
    except ValueError as e:
        print(f"捕获到异常: {e}")
        print(f"任务异常: {task.exception()}")

asyncio.run(main())

关键点

理解事件循环机制 - 这是异步编程的基础
正确使用async/await - 避免阻塞调用
合理管理并发 - 使用信号量限制并发数
妥善处理错误 - 确保程序的稳定性
注意资源清理 - 防止资源泄漏

ssh: connect to host github.com port 22: Connection timed out

Fri, 29 Aug 2025 10:24:50 +0800

github仓库可以网页打开，但git 无法使用

git -c diff.mnemonicprefix=false -c core.quotepath=false --no-optional-locks fetch --tags origin
ssh: connect to host github.com port 22: Connection timed out
fatal: Could not read from remote repository.
Please make sure you have the correct access rights
and the repository exists.

方法1

VPN用户，查看git 全局代理

 git config --global -l

我之前是配置了htts.proxy没用，发现加上http.proxy就好了，或者2者全加

git config --global http.proxy http://127.0.0.1:7897

方法2：

修改ssh配置文件，无则新建

windows

C:\Users\<your name>\.ssh\config

linux

~/.ssh/config

Host github.com
  Hostname ssh.github.com
  User git
  Port 443
  PreferredAuthentications publickey
  IdentityFile ~/.ssh/id_rsa

此处别忘了密钥要加到github里去，不会另找教程

添加完后使用命令测试

  ssh -T git@ssh.github.com -p 443

网上讲的方法2，对我无效，使用方法1后好了，但另一个仓库不行，再加上方法2后都可以访问。严重怀疑是github偷摸更新了啥，前些天一直都好好的

Linux 逻辑卷动态扩容

Tue, 03 Jun 2025 17:46:42 +0800

起因：盘满了，发现分区只有100G，但是物理盘有空间2T

用卷组中所有可用空间来扩展逻辑卷

lvextend -l +100%FREE /dev/mapper/ubuntu--vg-ubuntu--lv

对于ext4文件系统

resize2fs /dev/mapper/ubuntu--vg-ubuntu--lv

对于 xfs 文件系统

xfs_growfs /dev/mapper/ubuntu--vg-ubuntu--lv

adb connect 报 10061 解决方法

Thu, 24 Apr 2025 17:16:46 +0800

先用有线连接

adb tcpip 5555

adb connect 手机ip:5555

拔线

另外，别忘在开发者模式里面开启“无线调试”

Windows 防火墙怎么开启端口

Fri, 14 Feb 2025 17:43:05 +0800

以Windows11为例

此电脑-右键-属性- 隐私和安全性-Windows安全中心 (你也可以直接在开始菜单的搜索框里搜"安全中心")

右键添加入站规则，类型选端口，后面填入端口后就所有下一步就行了

Docker 使用代理拉镜像

Fri, 14 Feb 2025 17:16:27 +0800

DockerHub 访问非常慢，国内镜像源封杀干净，拉个镜像很费劲。

我的Windows电脑可以魔幻上网，所以把linux 中的docker添加代理，通过windows上网来拉取镜像

操作步骤

首先，确保存在用于存放 Docker 配置的目录。如果没有，请创建它。

sudo mkdir -p /etc/systemd/system/docker.service.d

创建或编辑代理配置文件：

接下来，创建或编辑一个名为 http-proxy.conf 的文件在这个目录下，并添加你的代理服务器信息。

sudo nano /etc/systemd/system/docker.service.d/http-proxy.conf

在文件中输入如下内容，记得替换为你实际的代理服务器地址和端口：

[Service] 
Environment="HTTP_PROXY=http://proxy.example.com:80/"
Environment="HTTPS_PROXY=https://proxy.example.com:443/" 
Environment="NO_PROXY=localhost,127.0.0.1"

如果你的代理服务器需要认证，可以将用户名和密码包含在URL中，例如：

Environment="HTTP_PROXY=http://username:password@proxy.example.com:80/"

重新加载 systemd 并重启 Docker 服务：

完成上述步骤后，运行以下命令以使更改生效：

sudo systemctl daemon-reload
sudo systemctl restart docker

验证设置：

最后，你可以通过检查 Docker 服务的状态来确认代理是否已正确应用：

systemctl show --property=Environment docke

这应该会输出你之前设置的环境变量。

然后docker pull 就会经过代理去拉取镜像

然后它就走了我的windows网络了

注意你的魔幻上网工具，记得开启“允许局域网连接”，把代理的端口号允许通过防火墙，参考 Windows如何开启端口

go 远程调试

Thu, 09 Jan 2025 17:30:23 +0800

服务器上

go install github.com/go-delve/delve/cmd/dlv@latest

本地，使用GoLand

服务器上也放一份源码，在源码目录，编译

go build -gcflags "all=-N -l"

两种启动方式

一。直接从本地启动

1. 服务器运行

dlv debug --headless --listen=:2345 --api-version=2 --accept-multiclient

2. goland 直接debug，注意提前打好断点

缺点，程序带不了参数

二。从服务器启动

1.运行程序，等待连接调试器

dlv --listen=:2345 --headless=true --api-version=2 --accept-multiclient exec ./demo.exe 参数1 参数2

2. 同上，打好断点，直接debug，启动参数也有了

3.如果自己的参数有 - ，如 demo.exe -f hello -b 200 之类的，在demo.exe 后加 -- 再接带-参数，否则会报错

博观网

Ubuntu中设置网络IP地址4种方法

方法1：使用netplan（Ubuntu 17.10及以后版本）

设置静态IP

设置为DHCP

方法2：使用NetworkManager（桌面版常用）

图形界面设置

命令行设置

方法3：传统ifupdown方法（较老版本）

设置静态IP

设置为DHCP

方法4：使用ip命令（临时设置）

临时设置静态IP

临时设置为DHCP

验证配置

注意事项

Python代码加密的四种方案

方案一：代码混淆

1. 手动混淆

2. 自动混淆工具

方案二：编译成字节码

1. 手动编译并分发 .pyc

方案三：打包成可执行文件

方案四：编译成原生代码

总结与对比

最终建议

Docker 与 Docker Compose 安装指南：使用代理加速下载

Docker 与 Docker Compose 安装指南：使用代理加速下载

第一部分：Docker 安装

标准安装步骤

使用代理加速下载

验证安装

第二部分：Docker Compose 安装

标准安装步骤

使用代理加速下载

第三部分：配置 Docker 使用代理拉取镜像

第四部分：常见问题解决

Python asyncio 用法深度解析

引言

一、核心概念解析

1.1 事件循环 (Event Loop)

1.2 协程 (Coroutine)

1.3 Future 和 Task

二、基本用法详解

2.1 并发执行多个任务

2.2 超时处理

2.3 同步原语

三、高级用法

3.1 队列 (Queue)

3.2 事件 (Event) 和条件 (Condition)

3.3 子进程管理

四、实际应用示例

4.1 并发HTTP请求

4.2 WebSocket客户端

五、注意事项与最佳实践

5.1 避免阻塞事件循环

5.2 正确的错误处理

5.3 资源清理

5.4 调试技巧

六、性能优化建议

6.1 限制并发数

6.2 使用连接池

6.3 批量操作

七、常见问题解答

Q1: 如何从同步代码调用异步函数？

Q2: 如何正确取消任务？

Q3: 如何处理任务中的异常？

关键点

ssh: connect to host github.com port 22: Connection timed out

Linux 逻辑卷动态扩容

adb connect 报 10061 解决方法

Windows 防火墙怎么开启端口

Docker 使用代理拉镜像

go 远程调试

两种启动方式

1. 手动编译并分发 `.pyc`