如何在pywin32中使用定时器

Question

我希望使用 Python 和 pywin32 计时器模块每 1 秒运行一次日志记录功能。这是我的代码

import timer    
i = 0
def logging():
    global i
    print(str(i))
    i += 1

timer.set_timer(1000, logging)

但是我在屏幕上看不到任何输出。我不确定我在这里做错了什么。

Answer 1

参见timer_demo.py（转载如下）。

pywin32

计时器基于 Windows 消息，因此需要实现消息循环。

# -*- Mode: Python; tab-width: 4 -*-
#

# This module, and the timer.pyd core timer support, were written by
# Sam Rushing ([email protected])

import timer
import time

# Timers are based on Windows messages.  So we need
# to do the event-loop thing!
import win32event, win32gui

# glork holds a simple counter for us.

class glork:

    def __init__ (self, delay=1000, max=10):
        self.x = 0
        self.max = max
        self.id = timer.set_timer (delay, self.increment)
        # Could use the threading module, but this is
        # a win32 extension test after all! :-)
        self.event = win32event.CreateEvent(None, 0, 0, None)

    def increment (self, id, time):
        print('x = %d' % self.x)
        self.x = self.x + 1
        # if we've reached the max count,
        # kill off the timer.
        if self.x > self.max:
            # we could have used 'self.id' here, too
            timer.kill_timer (id)
            win32event.SetEvent(self.event)

# create a counter that will count from '1' thru '10', incrementing
# once a second, and then stop.

def demo (delay=1000, stop=10):
    g = glork(delay, stop)
    # Timers are message based - so we need
    # To run a message loop while waiting for our timers
    # to expire.
    start_time = time.time()
    while 1:
        # We can't simply give a timeout of 30 seconds, as
        # we may continouusly be recieving other input messages,
        # and therefore never expire.
        rc = win32event.MsgWaitForMultipleObjects(
                (g.event,), # list of objects
                0, # wait all
                500,  # timeout
                win32event.QS_ALLEVENTS, # type of input
                )
        if rc == win32event.WAIT_OBJECT_0:
            # Event signalled.
            break
        elif rc == win32event.WAIT_OBJECT_0+1:
            # Message waiting.
            if win32gui.PumpWaitingMessages():
                raise RuntimeError("We got an unexpected WM_QUIT message!")
        else:
            # This wait timed-out.
            if time.time()-start_time > 30:
                raise RuntimeError("We timed out waiting for the timers to expire!")

if __name__=='__main__':
    demo()

输出（行间延迟 1 秒）：

x = 0
x = 1
x = 2
x = 3
x = 4
x = 5
x = 6
x = 7
x = 8
x = 9
x = 10

永远重要的最小例子：

import timer
import win32gui
import win32event

i = 0
def logging(timer_id, current_time):
    global i
    print(i)
    i += 1

timer.set_timer(1000, logging)

# The message loop waits for a windows message,
# then dispatches it.  The callback won't run
# without this processing.
while True:
    rc = win32event.MsgWaitForMultipleObjects((), False, win32event.INFINITE, win32event.QS_ALLEVENTS)
    if rc == win32event.WAIT_OBJECT_0:
        win32gui.PumpWaitingMessages()

如果你只想要一个后台计数器，你可以只使用一个线程：

import threading
import time

def logging():
    i = 0
    while True:
        time.sleep(1)
        print(i)
        i += 1

threading.Thread(target=logging).start()

# For some foreground activity:
while True:
    print('.', end='', flush=True)
    time.sleep(.2)

输出：

.....0
.....1
.....2      # etc...