def read_large_file(file_path): """ Generator function to read a large file line by line. """ with open(file_path, 'r') as file: for line in file: yield line
使用以下方法使用大文本中的数据
next 方法。调用生成器函数(read_large_file)，会返回一个 Generator 对象，通过 next() 方法会迭代调用生成器的下一个值（yield 表达式的值）
file_path = 'large_file.txt' # next 方法： 首先 line = read_large_file(file_path)
next(line) # 返回第一行 next(line) # 返回第二行，以此类推可以读取所有行
for 循环。调用生成器函数返回一个生成器对象，这个对象实现了迭代器协议。
def read_large_file(file_path): """ Generator function to read a large file line by line. """ with open(file_path, 'r') as file: for line in file: yield line # Usage example file_path = 'large_file.txt' for line in read_large_file(file_path): print(line.strip())
分批读取大文件中的数据
在处理大文件的过程中，如果需要批量多行读取文件内容，参考以下代码
def read_file_in_chunks(file_path, chunk_size=1024): """ Generator function to read a file in chunks. """ with open(file_path, 'r') as file: while True: chunk = file.readlines(chunk_size) if not chunk: break for line in chunk: yield line # Usage example file_path = 'large_file.txt' for line in read_file_in_chunks(file_path): print(line.strip())
tasks: - name: Install nginx only if nginx_install is true apt: name: nginx state: present when: nginx_install
- name: Install nginx on Debian or Ubuntu apt: name: nginx state: present when: ansible_facts['os_family'] == 'Debian' or ansible_facts['os_family'] == 'Ubuntu'
- name: Ensure package is installed if it is in the list apt: name: "{{ item }}" state: present loop: "{{ packages_to_install }}" when: item in packages_to_install
- name: Run only if the key 'run_task' is present in mydict and its value is true debug: msg: "Running task" when: mydict.get('run_task', False)
--- - name: Change hostname based on inventory alias hosts: test_target1 become: yes tasks: - name: Set hostname from alias hostname: name: "{{ inventory_hostname }}"
注意事项： Ansible 使用字典顺序加载配置文件，如果在不同的配置文件中配置了 parent groups 和 child groups，那么定义 child groups 的配置要先用定义 parent groups 的文件加载，否则 Ansible 加载配置会报错： Unable to parse /path/to/source_of_parent_groups as an inventory source[4]
# 单行字符串 name: John Doe # 整数 age: 35 # 浮点数 height: 5.9 # 布尔值 is_student: false
多行字符串可以使用字面量样式（|）或折叠样式（>）：
# 字面量样式保留换行符 address: | 123 Main St Anytown, WW 12345 # 折叠样式将连续的行合并为一行 description: > This is a very long sentence that spans several lines in the YAML but will be rendered as a single line in the output.
name: nextcloud_aio_mastercontainer: Volume 名称必须是 nextcloud_aio_mastercontainer，否则会报错找不到卷 nextcloud_aio_mastercontainer: It seems like you did not give the mastercontainer volume the correct name? (The 'nextcloud_aio_mastercontainer' volume was not found.). Using a different name is not supported since the built-in backup solution will not work in that case!
创建 Prometheus Server 配置文件，如 /root/prometheus/prometheus.yml，内容如下 [1]
/data/prometheus/prometheus.yml
# my global config global: scrape_interval: 15s # Set the scrape interval to every 15 seconds. Default is every 1 minute. evaluation_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute. # scrape_timeout is set to the global default (10s). # Alertmanager configuration alerting: alertmanagers: - static_configs: - targets: # - alertmanager:9093 # Load rules once and periodically evaluate them according to the global 'evaluation_interval'. rule_files: # - "first_rules.yml" # - "second_rules.yml" # A scrape configuration containing exactly one endpoint to scrape: # Here it's Prometheus itself. scrape_configs: # The job name is added as a label `job=<job_name>` to any timeseries scraped from this config. - job_name: 'prometheus'
# metrics_path defaults to '/metrics' # scheme defaults to 'http'.
static_configs: - targets: ['localhost:9090']
使用 Docker 启动时挂载此文件，作为 Prometheus Server 的配置文件，之后需要修改配置，可以直接修改此文件。
如果需要在非异步环境中（例如 Django 试图函数） 运行以上异步代码，会报错： RuntimeError: There is no current event loop in thread 'Thread-1'。需要特殊处理，可以使用 asyncio.run() 函数来运行异步代码，它可以在非异步环境中创建一个新的事件循环并运行异步函数。
