我需要從 Twisted 代碼中的網(wǎng)絡(luò)接口接收原始數(shù)據(jù)包。數(shù)據(jù)包沒有正確的 IP 或 MAC 地址，也沒有有效的標頭，所以我需要原始的東西。我曾嘗試查看twisted.pair，但我無法弄清楚如何使用它來獲取原始界面。通常，我會使用scapy.all.sniff. 但是，這是阻塞的，所以我不能將它與 Twisted 一起使用。（我也不能使用scapy.all.sniff超時和忙循環(huán)，因為我不想丟失數(shù)據(jù)包。）一個可能的解決方案是scapy.all.sniff在一個線程中運行，并在我收到數(shù)據(jù)包時以某種方式回調(diào) Twisted。這似乎有點不雅（而且，我不知道該怎么做，因為我是 Twisted 初學(xué)者），但如果我找不到更好的方法，我可能會接受。
1 回答
GCT1015
TA貢獻1827條經(jīng)驗 獲得超4個贊
您可以運行分布式系統(tǒng)并通過中央排隊系統(tǒng)傳遞數(shù)據(jù)。采用 Unix 哲學(xué)并創(chuàng)建一個單一的應(yīng)用程序，它可以完成一些任務(wù)并且做得很好。創(chuàng)建一個嗅探數(shù)據(jù)包的應(yīng)用程序（您可以scapy在此處使用，因為如果您阻止任何內(nèi)容都無關(guān)緊要），然后將它們發(fā)送到隊列（RabitMQ、Redis、SQS 等）并讓另一個應(yīng)用程序處理來自隊列的數(shù)據(jù)包。這種方法應(yīng)該給你帶來最少的頭痛。
如果您需要在單個應(yīng)用程序中運行所有內(nèi)容，那么線程/多處理是唯一的選擇。但是有一些設(shè)計模式你會想要遵循。您還可以將以下代碼分解為單獨的函數(shù)并使用專用的排隊系統(tǒng)。
from threading import Thread
from time import sleep
from twisted.internet import defer, reactor
class Sniffer(Thread):
def __init__(self, _reactor, shared_queue):
super().__init__()
self.reactor = _reactor
self.shared_queue = shared_queue
def run(self):
"""
Sniffer logic here
"""
while True:
self.reactor.callFromThread(self.shared_queue.put, 'hello world')
sleep(5)
@defer.inlineCallbacks
def consume_from_queue(_id, _reactor, shared_queue):
item = yield shared_queue.get()
print(str(_id), item)
_reactor.callLater(0, consume_from_queue, _id, _reactor, shared_queue)
def main():
shared_queue = defer.DeferredQueue()
sniffer = Sniffer(reactor, shared_queue)
sniffer.daemon = True
sniffer.start()
workers = 4
for i in range(workers):
consume_from_queue(i+1, reactor, shared_queue)
reactor.run()
main()
該Sniffer班開始扭曲的控制之外。請注意sniffer.daemon = True，這是為了在主線程停止時線程將停止。如果它被設(shè)置為False（默認），那么只有當所有線程都結(jié)束時，應(yīng)用程序才會退出。根據(jù)手頭的任務(wù)，這可能總是也可能不總是可能的。如果您可以暫停嗅探以檢查線程事件，那么您也許能夠以更安全的方式停止線程。
self.reactor.callFromThread(self.shared_queue.put, 'hello world')是必要的，以便放入隊列的項目發(fā)生在主反應(yīng)器線程中，而不是Sniffer執(zhí)行的線程中。這樣做的主要好處是來自線程的消息會有某種同步（假設(shè)您計劃擴展到嗅探多個接口）。另外，我不確定DeferredQueue對象是否是線程安全的 :) 我像對待它們一樣對待它們。
由于在這種情況下 Twisted 不管理線程，因此開發(fā)人員管理至關(guān)重要。注意worker循環(huán)和consume_from_queue(i+1, reactor, shared_queue)。此循環(huán)確保只有所需數(shù)量的工作人員正在處理任務(wù)。在consume_from_queue()函數(shù)內(nèi)部，shared_queue.get()將等待（非阻塞），直到將一個項目放入隊列，打印該項目，然后安排另一個consume_from_queue().
添加回答
舉報
0/150
提交
取消