Python 用心跳(UDP包)探测不活动主机
Python 用心跳(UDP包)探测不活动主机
计算机周期性的发送一个代表心跳的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'
- whileTrue:
- 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) inself.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):
- whileself.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:
- whileTrue:
- 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( )
作为备选方案,线面给出异步的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) inself.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)
评论暂时关闭