我的任务是编写一个 Python 脚本,生成指定数量的二维码并将每个二维码保存在 PDF 文件的单独页面上。每个二维码都放置在一个模板上,该模板的顶部和底部均包含公司徽标。虽然我当前的代码可以运行,但生成 10,000 个二维码大约需要 200 秒。我被要求优化此流程,将时间缩短至 40 秒以下。如何实现更好的多线程来满足这个要求?
代码如下
import os
import qrcode
import time
import tempfile
from PIL import Image
from concurrent.futures import ThreadPoolExecutor, as_completed
from reportlab.pdfgen import canvas
def generate_qr_code(data, height_inches, width_inches, dpi=320):
height = height_inches * dpi
width = width_inches * dpi
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=15,
border=4,
)
qr.add_data(data)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
img = img.resize((width, height), Image.ANTIALIAS)
return img
def add_qr_to_template(template_path, qr_img, qr_position):
template_path = os.path.expanduser(template_path)
try:
template = Image.open(template_path)
except Exception as e:
print(f"Error opening template image: {e}")
return None
template.paste(qr_img, qr_position)
return template
def add_qr_to_pdf(pdf_canvas, qr_img, qr_position):
# Save the QR code image to a temporary file
with tempfile.NamedTemporaryFile(delete=False, suffix='.png') as temp_file:
qr_img.save(temp_file, format='PNG')
temp_file_path = temp_file.name
# Draw the image from the temporary file
pdf_canvas.drawImage(temp_file_path, 0, 0, width=289.2, height=418.50)
pdf_canvas.showPage()
# Clean up the temporary file
os.remove(temp_file_path)
def generate_and_add_qr_to_pdf(pdf_canvas, data, qr_position, template_path):
qr_img = generate_qr_code(data, 2, 2)
template = add_qr_to_template(template_path, qr_img, qr_position)
if template:
add_qr_to_pdf(pdf_canvas, template, qr_position)
def main():
num_qr_codes = int(input("Enter the number of QR codes to generate: "))
data = "examplesite.com"
template_path = "QrGenerator/company_logo.png"
qr_position = (2, 252) # Adjust this to fit your PDF
pdf_path = "QrGenerator/qr_codes.pdf"
custom_size=(289.29,418.50)
c = canvas.Canvas(pdf_path, pagesize=custom_size)
start_time = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
futures = []
for i in range(num_qr_codes):
futures.append(executor.submit(generate_and_add_qr_to_pdf, c, data, qr_position, template_path))
# Wait for all threads to complete
for future in as_completed(futures):
future.result()
# Save the PDF
c.save()
end_time = time.time()
total_time = end_time - start_time
print(f"Time taken to generate and save {num_qr_codes} QR codes into PDF: {total_time:.2f} seconds")
print(f"PDF saved as {pdf_path}")
if __name__ == "__main__":
main()
一些建议:
首先,创建位图(QR 码)并调整其大小需要占用 CPU 资源,因此多处理比多线程更适合。其次,我不相信同时将 PNG 文件写入磁盘会给您带来多大好处。根据您的硬件,这甚至可能会损害性能。
如果每个图像的实际数据是动态生成的,那么您可以通过使用生成器函数或生成器表达式来生成数据来节省一些时间和内存。然后,我将使用
multiprocessing.pool.Pool.imap
函数根据池大小和创建的图像数量指定合适的 chunksize 参数。使用 concurrent.futures.ThreadPoolExecutor.submit
或或多或少等效的 multiprocessing.pool.Pool.apply_async
方法不允许任务被 chunked 导致对池的内部队列进行更多的写入和读取操作。分块提供更大但更少的读/写队列操作,并且在提交的任务数量“很大”时强烈建议使用分块。一旦从工作函数返回结果,所使用的 imap
方法就会按照任务提交顺序返回已提交任务的结果。
请注意,在当前代码中未定义提交任务的完成顺序。这并不重要,因为您为每个 PDF 页面生成相同的 PNG 文件。但是,如果在“真实代码”中,每个页面都需要一个特定的 PNG 文件,那么您的代码将无法工作,因为提交到池的第二个任务可能会先完成,并且您最终会将用于第二个页面的 PNG 图像添加到相反,第一页。如果 PNG 文件的添加顺序确实并不重要,则在以下代码中将对方法
imap
的调用替换为 imap_unordered
以获得更好的性能。
以下是如何使用多重处理来生成 PNG 图像并将它们连续添加到 PDF 文件的一般思路:
import qrcode
from PIL import Image
import time
from multiprocessing import Pool, cpu_count
from functools import partial
def generate_qr_code(data, height_inches, width_inches, dpi=320):
height = height_inches * dpi
width = width_inches * dpi
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=15,
border=4,
)
qr.add_data(data)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
img = img.resize((width, height))
return img
def main():
N_QR_CODES = 1_000 # Just for demo purposes
DATA = 'https://examplesite.com'
def generate_data():
"""For demo purposes generate identical data."""
for _ in range(N_QR_CODES):
yield DATA
def compute_chunksize(iterable_size, pool_size):
chunksize, remainder = divmod(iterable_size, 4 * pool_size)
if remainder:
chunksize += 1
return chunksize
t = time.time()
pool_size = cpu_count()
iterable_size = N_QR_CODES
chunksize = compute_chunksize(iterable_size, pool_size)
worker = partial(generate_qr_code, height_inches=2, width_inches=2)
with Pool(pool_size) as pool:
# Use a generator or generator function if you are able to:
for img in pool.imap(worker, generate_data(), chunksize=chunksize):
# Write out image or add to PDF
...
print(time.time() - t)
if __name__ == "__main__":
main()