計算機週期性的傳送一個代表心跳的UDP包到伺服器，伺服器跟蹤每臺計算機在上次傳送心跳之後盡力的時間並報告那些沉默時間太長的計算機。

客戶端程式：HeartbeatClient.py

""" 心跳客戶端,週期性的傳送 UDP包 """
import socket, time
SERVER_IP = '192.168.0.15'; SERVER_PORT = 43278; BEAT_PERIOD = 5
print 'Sending heartbeat to IP %s , port %d' % (SERVER_IP, SERVER_PORT)
print 'press Ctrl-C to stop'
while True:
    hbSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    hbSocket.sendto('PyHB', (SERVER_IP, SERVER_PORT))
    if _ _debug_ _:
        print 'Time: %s' % time.ctime( )
    time.sleep(BEAT_PERIOD)
 

伺服器程式接受ing跟蹤“心跳”，她執行的計算機的地址必須和“客戶端”程式中的 SERVER_IP一致。伺服器必須支援併發，因為來自不同的計算機的心跳可能會同時到達。一個伺服器有兩種方法支援併發：多執行緒和非同步操作。下面是一個多執行緒的ThreadbearServer.py，只使用了python標準庫中的模組：

""" 多執行緒 heartbeat 伺服器"""
import socket, threading, time
UDP_PORT = 43278; CHECK_PERIOD = 20; CHECK_TIMEOUT = 15
class Heartbeats(dict):
    """ Manage shared heartbeats dictionary with thread locking """
    def _ _init_ _(self):
        super(Heartbeats, self)._ _init_ _( )
        self._lock = threading.Lock( )
    def _ _setitem_ _(self, key, value):
        """ Create or update the dictionary entry for a client """
        self._lock.acquire( )
        try:
            super(Heartbeats, self)._ _setitem_ _(key, value)
        finally:
            self._lock.release( )
    def getSilent(self):
        """ Return a list of clients with heartbeat older than CHECK_TIMEOUT """
        limit = time.time( ) - CHECK_TIMEOUT
        self._lock.acquire( )
        try:
            silent = [ip for (ip, ipTime) in self.items( ) if ipTime < limit]
        finally:
            self._lock.release( )
        return silent
class Receiver(threading.Thread):
    """ Receive UDP packets and log them in the heartbeats dictionary """
    def _ _init_ _(self, goOnEvent, heartbeats):
        super(Receiver, self)._ _init_ _( )
        self.goOnEvent = goOnEvent
        self.heartbeats = heartbeats
        self.recSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.recSocket.settimeout(CHECK_TIMEOUT)
        self.recSocket.bind(('', UDP_PORT))
    def run(self):
        while self.goOnEvent.isSet( ):
            try:
                data, addr = self.recSocket.recvfrom(5)
                if data == 'PyHB':
                    self.heartbeats[addr[0]] = time.time( )
            except socket.timeout:
                pass
def main(num_receivers=3):
    receiverEvent = threading.Event( )
    receiverEvent.set( )
    heartbeats = Heartbeats( )
    receivers = [  ]
    for i in range(num_receivers):
        receiver = Receiver(goOnEvent=receiverEvent, heartbeats=heartbeats)
        receiver.start( )
        receivers.append(receiver)
    print 'Threaded heartbeat server listening on port %d' % UDP_PORT
    print 'press Ctrl-C to stop'
    try:
        while True:
            silent = heartbeats.getSilent( )
            print 'Silent clients: %s' % silent
            time.sleep(CHECK_PERIOD)
    except KeyboardInterrupt:
        print 'Exiting, please wait...'
        receiverEvent.clear( )
        for receiver in receivers:
            receiver.join( )
        print 'Finished.'
if _ _name_ _ == '_ _main_ _':
    main( )
 

NB：在執行該程式時可能出現“ socket.error: [Errno 98] Address already in use”（linux下） 或 “socket.error: [Errno 10048] 通常每個套接字地址（協議/網路地址/埠）只允許使用一次”（windows下），解決辦法參見博文：

作為備選方案，線面給出非同步的AsyBeatserver.py程式，這個程式接住了強大的twisted的力量：

import time
from twisted.application import internet, service
from twisted.internet import protocol
from twisted.python import log
UDP_PORT = 43278; CHECK_PERIOD = 20; CHECK_TIMEOUT = 15
class Receiver(protocol.DatagramProtocol):
    """ Receive UDP packets and log them in the "client"s dictionary """
    def datagramReceived(self, data, (ip, port)):
        if data == 'PyHB':
            self.callback(ip)
class DetectorService(internet.TimerService):
    """ Detect clients not sending heartbeats for too long """
    def _ _init_ _(self):
        internet.TimerService._ _init_ _(self, CHECK_PERIOD, self.detect)
        self.beats = {  }
    def update(self, ip):
        self.beats[ip] = time.time( )
    def detect(self):
        """ Log a list of clients with heartbeat older than CHECK_TIMEOUT """
        limit = time.time( ) - CHECK_TIMEOUT
        silent = [ip for (ip, ipTime) in self.beats.items( ) if ipTime < limit]
        log.msg('Silent clients: %s' % silent)
application = service.Application('Heartbeat')
# define and link the silent clients' detector service
detectorSvc = DetectorService( )
detectorSvc.setServiceParent(application)
# create an instance of the Receiver protocol, and give it the callback
receiver = Receiver( )
receiver.callback = detectorSvc.update
# define and link the UDP server service, passing the receiver in
udpServer = internet.UDPServer(UDP_PORT, receiver)
udpServer.setServiceParent(application)
# each service is started automatically by Twisted at launch time
log.msg('Asynchronous heartbeat server listening on port %d\n'
    'press Ctrl-C to stop\n' % UDP_PORT)