NameError: name 'x' is not defined 是 Python 中常見的錯誤之一，通常表示你嘗試訪問一個尚未定義的變數或函式。特別是全域性名稱未定義時，意味著你在使用某個全域性變數或函式時，Python 在當前名稱空間中找不到該名稱。
1、問題背景
在使用 Python 時，如果遇到了 NameError: global name 'control_queue' is not defined 的錯誤，通常是因為嘗試訪問一個尚未定義的全域性變數。這種情況常見於多執行緒或多程序程式設計中，因為不同執行緒或程序可能無法訪問彼此的區域性變數。
在下面示例的程式碼中，control_queue 變數在函式 halt_listener 中被使用，但是它並沒有在函式外部定義。這就會導致 NameError: global name 'control_queue' is not defined 的錯誤。
2、解決方案
要解決此問題，需要在函式外部定義 control_queue 變數，或者將 control_queue 傳遞給 halt_listener 函式作為引數。
2.1 將 control_queue 定義為全域性變數
import multiprocessing
import time
from threading import Thread
class test_imports:#Test classes remove
alive = {'import_1': True, 'import_2': True};
def __init__(self, control_queue):
self.control_queue = control_queue
def halt_listener(self, thread_Name, kill_command):
while True:
print ("Checking queue for kill")
isAlive = self.control_queue.get()
print ("isAlive", isAlive)
if isAlive == kill_command:
print ("kill listener triggered")
self.alive[thread_Name] = False;
return
def import_1(self, thread_Number):
print ("Import_1 number %d started") % thread_Number
halt = test_imports()
t = Thread(target=halt.halt_listener, args=(control_Queue, 'import_1', 't1kill'))
count = 0
t.run()
global alive
run = test_imports.alive['import_1'];
while run:
print ("Thread type 1 number %d run count %d") % (thread_Number, count)
count = count + 1
print ("Test Import_1 ", run)
run = self.alive['import_1'];
print ("Killing thread type 1 number %d") % thread_Number
def import_2(self, thread_number):
print ("Import_2 number %d started") % thread_number
count = 1
while True:
alive = self.control_queue.get()
count = count + 1
if alive == 't2kill':
print ("Killing thread type 2 number %d") % thread_number
return
else:
print ("Thread type 2 number %d run count %d") % (thread_number, count)
class worker_manager:
def __init__(self):
self.children = {}
def generate(self, control_queue, threadName, runNum):
i = test_imports(control_queue)
if threadName == 'one':
print ("Starting import_1 number %d") % runNum
p = multiprocessing.Process(target=i.import_1, args=(runNum,))
self.children[threadName] = p
p.start()
elif threadName == 'two':
print ("Starting import_2 number %d") % runNum
p = multiprocessing.Process(target=i.import_2, args=(runNum,))
self.children[threadName] = p
p.start()
elif threadName == 'three':
p = multiprocessing.Process(target=i.import_1, args=(runNum,))
print ("Starting import_1 number %d") % runNum
p2 = multiprocessing.Process(target=i.import_2, args=(runNum,))
print ("Starting import_2 number %d") % runNum
self.children[threadName] = p
self.children[threadName] = p2
p.start()
p2.start()
else:
print ("Not a valid choice choose one two or three")
def terminate(self, threadName):
self.children[threadName].join
if __name__ == '__main__':
# Establish communication queues
control_queue = multiprocessing.Queue()
manager = worker_manager()
runNum = int(raw_input("Enter a number: "))
threadNum = int(raw_input("Enter number of threads: "))
threadName = raw_input("Enter number: ")
thread_Count = 0
print ("Starting threads")
for i in range(threadNum):
manager.generate(control_queue, threadName, i)
thread_Count = thread_Count + 1
time.sleep(runNum)#let threads do their thing
print ("Terminating threads")
for i in range(thread_Count):
control_queue.put("t1kill")
control_queue.put("t2kill")
manager.terminate(threadName)
2.2 將 control_queue 傳遞給 halt_listener 函式作為引數
import multiprocessing
import time
from threading import Thread
class test_imports:#Test classes remove
alive = {'import_1': True, 'import_2': True};
def halt_listener(self, control_queue, thread_Name, kill_command):
while True:
print ("Checking queue for kill")
isAlive = control_queue.get()
print ("isAlive", isAlive)
if isAlive == kill_command:
print ("kill listener triggered")
self.alive[thread_Name] = False;
return
def import_1(self, control_queue, thread_Number):
print ("Import_1 number %d started") % thread_Number
halt = test_imports()
t = Thread(target=halt.halt_listener, args=(control_queue, 'import_1', 't1kill'))
count = 0
t.run()
global alive
run = test_imports.alive['import_1'];
while run:
print ("Thread type 1 number %d run count %d") % (thread_Number, count)
count = count + 1
print ("Test Import_1 ", run)
run = self.alive['import_1'];
print ("Killing thread type 1 number %d") % thread_Number
def import_2(self, control_queue, thread_number):
print ("Import_2 number %d started") % thread_number
count = 1
while True:
alive = control_queue.get()
count = count + 1
if alive == 't2kill':
print ("Killing thread type 2 number %d") % thread_number
return
else:
print ("Thread type 2 number %d run count %d") % (thread_number, count)
class worker_manager:
def __init__(self):
self.children = {}
def generate(self, control_queue, threadName, runNum):
i = test_imports()
if threadName == 'one':
print ("Starting import_1 number %d") % runNum
p = multiprocessing.Process(target=i.import_1, args=(control_queue, runNum))
self.children[threadName] = p
p.start()
elif threadName == 'two':
print ("Starting import_2 number %d") % runNum
p = multiprocessing.Process(target=i.import_2, args=(control_queue, runNum))
self.children[threadName] = p
p.start()
elif threadName == 'three':
p = multiprocessing.Process(target=i.import_1, args=(control_queue, runNum))
print ("Starting import_1 number %d") % runNum
p2 = multiprocessing.Process(target=i.import_2, args=(control_queue, runNum))
print ("Starting import_2 number %d") % runNum
self.children[threadName] = p
self.children[threadName] = p2
p.start()
p2.start()
else:
print ("Not a valid choice choose one two or three")
def terminate(self, threadName):
self.children[threadName].join
if __name__ == '__main__':
# Establish communication queues
control_queue = multiprocessing.Queue()
manager = worker_manager()
runNum = int(raw_input("Enter a number: "))
threadNum = int(raw_input("Enter number of threads: "))
threadName = raw_input("Enter number: ")
thread_Count = 0
print ("Starting threads")
for i in range(threadNum):
manager.generate
最後總結
NameError 的出現通常意味著 Python 找不到你引用的變數或函式。爲了避免這個錯誤，你可以：
確保變數或函式已定義。
在函式內部修改全域性變數時，使用
global宣告。在呼叫之前定義函式。
匯入需要的模組。
檢查拼寫。
理解作用域的限制。
這些實踐可以幫助你更有效地避免和解決 NameError。
