Python 지연 루프를 실행하는 가장 고통스러운 방법입니다.

Question

이벤트 구동 괄호가 있는데 스팸 방지 기능을 구현하려고합니다. 나는 응용 프로그램의 나머지 부분을 차단하지 않고 일정 기간 동안 심하게 행동하는 사용자를 침묵 시키려고합니다.

다음은 작동하지 않습니다 무엇 :

는

if user_behaving_badly(): 
    ban(user) 
    time.sleep(penalty_duration) # Bad! Blocks the entire application! 
    unban(user) 

이상적으로, user_behaving_badly()가 true 인 경우, 나는 아무것도하지 않는 새로운 스레드를 시작하지만 사용자를 금지 한 후 잠시 동안 잠을하려는의 접근을 허용 사용자, 스레드가 사라집니다.

According to this 나는 다음과 같은 사용하여 내 목표를 달성 할 수

if user_behaving_badly(): 
    thread.start_new_thread(banSleepUnban, (user, penalty)) 

"단순이"보통 "좋은"의 지표이며,이 꽤 간단하지만 내가 스레드에 대해 들어 본 모든 것을 말했다 그들은 예상치 못한 방식으로 당신을 물 수 있습니다. 내 질문 : 이 응용 프로그램의 나머지 부분을 차단하지 않고 간단한 지연 루프를 실행하는보다 나은 방법 있는가?

Answer 1

사용과 같은 스레딩 timer object :

t = threading.Timer(30.0, unban) 
t.start() # after 30 seconds, unban will be run 

스레드에서 실행 차단 해제 만. 대신 각각의 금지에 대한 스레드를 시작하는

Answer 2

이것은 언어에 구애받지 않지만, 물건을 추적하는 스레드라고 생각합니다. 스레드는 테이블에 "username"및 "banned_until"과 같은 데이터 구조를 유지합니다. 스레드는 항상 테이블을 검사하는 백그라운드에서 실행되고, banned_until이 만료되면 사용자를 차단 해제합니다. 다른 스레드는 정상적으로 진행됩니다.

Answer 3

왜 스레드가 필요합니까? 당신이 GUI를 사용하는 경우

do_something(user): 
    if(good_user(user)): 
    # do it 
    else 
    # don't 

good_user(): 
    if(is_user_baned(user)): 
    if(past_time_since_ban(user)): 
     user_good_user(user) 
    elif(is_user_bad()): 
    ban_user() 

ban_user(user): 
    # add a user/start time to a hash 

is_user_banned() 
    # check hash 
    # could check if expired now too, or do it seperately if you care about it 

is_user_bad() 
    # check params or set more values in a hash 

Answer 4

,
대부분의 GUI 모듈은 타이머 기능이있는 추상화 할 수있는 모든 우웩 멀티 스레딩 물건, 아직도의 나머지 부분을 허용하지만, 주어진 시간 후 를 코드를 실행 실행될 코드.

예를 들어, Tkinter에는 'after'기능이 있습니다.

Answer 5

는, 우선 순위 큐에 금지를 넣어

이

이 코드는 두 구조가 빠르게 조속한 금지를 찾을 수있는 heapq를 유지 unbanning 수면과을 단일 스레드를 만료하고 사용자가 이름

import time 
import threading 
import heapq 

class Bans(): 
    def __init__(self): 
     self.lock = threading.Lock() 
     self.event = threading.Event() 
     self.heap = [] 
     self.dict = {} 
     self.thread = threading.thread(target=self.expiration_thread) 
     self.thread.setDaemon(True) 
     self.thread.start() 

    def ban_user(self, name, duration): 
     with self.lock: 
      now = time.time() 
      expiration = (now+duration) 
      heapq.heappush(self.heap, (expiration, user)) 
      self.dict[user] = expiration 
      self.event.set() 

    def is_user_banned(self, user): 
     with self.lock: 
      now = time.time() 
      return self.dict.get(user, None) > now 

    def expiration_thread(self): 
     while True: 
      self.event.wait() 
      with self.lock: 
       next, user = self.heap[0] 
       now = time.time() 
       duration = next-now 
      if duration > 0: 
       time.sleep(duration) 
      with self.lock: 
       if self.heap[0][0] = next: 
        heapq.heappop(self.heap) 
        del self.dict(user) 
       if not self.heap: 
        self.event.clear() 

에 의해 금지되며, 다음과 같이 사용되는 경우 딕셔너리가 가능한 신속하게 확인할 수 있도록 :

B = Bans() 
B.ban_user("phil", 30.0) 
B.is_user_banned("phil")