介紹
優化上一個挑戰中的計算器，支持使用多進程的方式對員工工資數據進行處理，以應對文件數據量很大的情況下，提高計算效率。

程序的執行過程如下，注意配置文件和輸入的員工數據文件需要你自己創建并填入數據，可以參考上述的內容示例：

$ ./calculator.py -c test.cfg -d user.csv -o gongzi.csv

執行成功不需要輸出信息到屏幕，執行失敗或有異常出現則將錯誤信息輸出到屏幕。

需要注意的是必須包含下列的處理方式：

啟動三個進程，使用進程 1 讀取員工工資數據，使用進程 2 計算個稅及社保，使用進程 3 將數據寫入到輸出的工資單數據文件中。
三個進程負責不同的工作，進程之間使用某種機制進行通信。
目標
完成任務需要達成的目標：

程序存放的位置 /home/shiyanlou/calculator.py
程序必須采用多進程的方式處理員工工資數據，并保證進程間能夠同步
提示語
下述實現方案僅供參考，會涉及到先前實驗中學習到的知識點，如果自己對程序有足夠的理解也可以不按照下述提示編寫

基于 multiprocessing 模塊實現多進程

基于 Queue 實現進程間通信

實現完成后，可以考慮是否可以在計算環節的進程2實現為一個進程池？

實現方案可以考慮定義 queue1 和 queue2，實現三個進程如下：

進程 1：從用戶文件中讀取數據，然后得到一個列表 data，第一項是用戶 ID，第二項是稅前工資，然后使用 queue1.put(data)。
進程 2：queue1.get() 得到列表 data，第一項是用戶 ID，第二項是稅前工資，然后計算后生成新的列表。 newdata，包含社保，個稅，稅后工資等數據，然后使用 queue2.put(newdata)。
進程 3：queue2.get() 得到列表 newdata，包含用戶 ID，稅前工資，社保，個稅，稅后工資等數據，然后寫入文件。
最后，因為后續的挑戰將會用到現在寫的代碼，請使用 下載代碼 保存到本地或者提交到自己的 Github

# -*- coding: utf-8 -*- import sys import csv from collections import namedtuple import queue from multiprocessing import Queue, Process# 稅率表條目類，該類由 namedtuple 動態創建，代表一個命名元組 IncomeTaxQuickLookupItem = namedtuple('IncomeTaxQuickLookupItem',['start_point', 'tax_rate', 'quick_subtractor'] )# 起征點常量 INCOME_TAX_START_POINT = 5000# 稅率表，里面的元素類型為前面創建的 IncomeTaxQuickLookupItem INCOME_TAX_QUICK_LOOKUP_TABLE = [IncomeTaxQuickLookupItem(80000, 0.45, 15160),IncomeTaxQuickLookupItem(55000, 0.35, 7160),IncomeTaxQuickLookupItem(35000, 0.30, 4410),IncomeTaxQuickLookupItem(25000, 0.25, 2660),IncomeTaxQuickLookupItem(12000, 0.2, 1410),IncomeTaxQuickLookupItem(3000, 0.1, 210),IncomeTaxQuickLookupItem(0, 0.03, 0) ]class Args(object):"""命令行參數處理類"""def __init__(self):# 保存命令行參數列表self.args = sys.argv[1:]def _value_after_option(self, option):"""內部函數，用來獲取跟在選項后面的值"""try:# 獲得選項位置index = self.args.index(option)# 下一位置即為選項值return self.args[index + 1]except (ValueError, IndexError):print('Parameter Error')exit()@propertydef config_path(self):"""配置文件路徑"""return self._value_after_option('-c')@propertydef userdata_path(self):"""用戶工資文件路徑"""return self._value_after_option('-d')@propertydef export_path(self):"""稅后工資文件路徑"""return self._value_after_option('-o')# 創建一個全局參數類對象供后續使用 args = Args()class Config(object):"""配置文件處理類"""def __init__(self):# 讀取配置文件self.config = self._read_config()def _read_config(self):"""內部函數，用來讀取配置文件中的配置項"""config = {}with open(args.config_path) as f:# 依次讀取配置文件里的每一行并解析得到配置項名稱和值for line in f.readlines():key, value = line.strip().split('=')try:# 去掉前后可能出現的空格config[key.strip()] = float(value.strip())except ValueError:print('Parameter Error')exit()return configdef _get_config(self, key):"""內部函數，用來獲得配置項的值"""try:return self.config[key]except KeyError:print('Config Error')exit()@propertydef social_insurance_baseline_low(self):"""獲取社保基數下限"""return self._get_config('JiShuL')@propertydef social_insurance_baseline_high(self):"""獲取社保基數上限"""return self._get_config('JiShuH')@propertydef social_insurance_total_rate(self):"""獲取社保總費率"""return sum([self._get_config('YangLao'),self._get_config('YiLiao'),self._get_config('ShiYe'),self._get_config('GongShang'),self._get_config('ShengYu'),self._get_config('GongJiJin')])# 創建一個全局的配置文件處理對象供后續使用 config = Config()class UserData(Process):"""用戶工資文件處理進程"""def __init__(self, userdata_queue):super().__init__()# 用戶數據隊列self.userdata_queue = userdata_queuedef _read_users_data(self):"""內部函數，用來讀取用戶工資文件"""userdata = []with open(args.userdata_path) as f:# 依次讀取用戶工資文件中的每一行并解析得到用戶 ID 和工資for line in f.readlines():employee_id, income_string = line.strip().split(',')try:income = int(income_string)except ValueError:print('Parameter Error')exit()userdata.append((employee_id, income))return userdatadef run(self):"""進程入口方法"""# 從用戶數據文件依次讀取每條用戶數據并寫入到隊列for item in self._read_users_data():self.userdata_queue.put(item)class IncomeTaxCalculator(Process):"""稅后工資計算進程"""def __init__(self, userdata_queue, export_queue):super().__init__()# 用戶數據隊列self.userdata_queue = userdata_queue# 導出數據隊列self.export_queue = export_queue@staticmethoddef calc_social_insurance_money(income):"""計算應納稅額"""if income < config.social_insurance_baseline_low:return config.social_insurance_baseline_low * \config.social_insurance_total_rateelif income > config.social_insurance_baseline_high:return config.social_insurance_baseline_high * \config.social_insurance_total_rateelse:return income * config.social_insurance_total_rate@classmethoddef calc_income_tax_and_remain(cls, income):"""計算稅后工資"""# 計算社保金額social_insurance_money = cls.calc_social_insurance_money(income)# 計算應納稅額real_income = income - social_insurance_moneytaxable_part = real_income - INCOME_TAX_START_POINT# 從高到低判斷落入的稅率區間，如果找到則用該區間的參數計算納稅額并返回結果for item in INCOME_TAX_QUICK_LOOKUP_TABLE:if taxable_part > item.start_point:tax = taxable_part * item.tax_rate - item.quick_subtractorreturn '{:.2f}'.format(tax), '{:.2f}'.format(real_income - tax)# 如果沒有落入任何區間，則返回 0return '0.00', '{:.2f}'.format(real_income)def calculate(self, employee_id, income):"""計算單個用戶的稅后工資"""# 計算社保金額social_insurance_money = '{:.2f}'.format(self.calc_social_insurance_money(income))# 計算稅后工資tax, remain = self.calc_income_tax_and_remain(income)return [employee_id, income, social_insurance_money, tax, remain]def run(self):"""進程入口方法"""# 從用戶數據隊列讀取用戶數據，計算用戶稅后工資，然后寫入到導出數據隊列while True:# 獲取下一個用戶數據try:# 超時時間為 1 秒，如果超時則認為沒有需要處理的數據，退出進程employee_id, income = self.userdata_queue.get(timeout=1)except queue.Empty:return# 計算稅后工資result = self.calculate(employee_id, income)# 將結果寫入到導出數據隊列self.export_queue.put(result)class IncomeTaxExporter(Process):"""稅后工資導出進程"""def __init__(self, export_queue):super().__init__()# 導出數據隊列self.export_queue = export_queue# 創建 CSV 寫入器self.file = open(args.export_path, 'w', newline='')self.writer = csv.writer(self.file)def run(self):"""進程入口方法"""# 從導出數據隊列讀取導出數據，寫入到導出文件中while True:# 獲取下一個導出數據try:# 超時時間為 1 秒，如果超時則認為沒有需要處理的數據，退出進程item = self.export_queue.get(timeout=1)except queue.Empty:# 退出時關閉文件self.file.close()return# 寫入到導出文件self.writer.writerow(item)if __name__ == '__main__':# 創建進程之間通信的隊列userdata_queue = Queue()export_queue = Queue()# 用戶數據進程userdata = UserData(userdata_queue)# 稅后工資計算進程calculator = IncomeTaxCalculator(userdata_queue, export_queue)# 稅后工資導出進程exporter = IncomeTaxExporter(export_queue)# 啟動進程userdata.start()calculator.start()exporter.start()# 等待所有進程結束userdata.join()calculator.join()exporter.join()

如果想將一個類直接改造成一個進程，可以直接將那個類直接繼承process，可以直接使用join、start等一些函數運行
Args和Config都不用單獨去實現

總結

以上是生活随笔為你收集整理的多进程工资计算器的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。