python实现的文件同步服务器实例

本文实例讲述了python实现的文件同步服务器。分享给大家供大家参考。具体实现方法如下：

服务端使用asyncore, 收到文件后保存到本地。

客户端使用pyinotify监视目录的变化 ，把变动的文件发送到服务端。

重点：

1. 使用structs打包发送文件的信息，服务端收到后，根据文件信息来接收客户端传送过来的文件。

2. 客户端使用多线程，pyinotify监视到文件变化，放到队列中，由另外一个线程发送。

上代码：

服务端：

# receive file from client and store them into file use asyncore.#
#/usr/bin/python
#coding: utf-8
import asyncore
import socket
from socket import errno
import logging
import time
import sys
import struct
import os
import fcntl
import threading
from rrd_graph import MakeGraph
try:
  import rrdtool
except (ImportError, ImportWarnning):
  print "Hope this information can help you:"
  print "Can not find pyinotify module in sys path, just run [apt-get install python-rrdtool] in ubuntu."
  sys.exit(1)
class RequestHandler(asyncore.dispatcher):
  def __init__(self, sock, map=None, chunk_size=1024):
    self.logger = logging.getLogger('%s-%s' % (self.__class__.__name__, str(sock.getsockname())))
    self.chunk_size = chunk_size
    asyncore.dispatcher.__init__(self,sock,map)
    self.data_to_write = list()
  def readable(self):
    #self.logger.debug("readable() called.")
    return True
  def writable(self):
    response = (not self.connected) or len(self.data_to_write)
    #self.logger.debug('writable() -> %s data length -> %s' % (response, len(self.data_to_write)))
    return response
  def handle_write(self):
    data = self.data_to_write.pop()
    #self.logger.debug("handle_write()->%s size: %s",data.rstrip('\r\n'),len(data))
    sent = self.send(data[:self.chunk_size])
    if sent < len(data):
      remaining = data[sent:]
      self.data_to_write.append(remaining)
  def handle_read(self):
    self.writen_size = 0
    nagios_perfdata = '../perfdata'
    head_packet_format = "!LL128s128sL"
    head_packet_size = struct.calcsize(head_packet_format)
    data = self.recv(head_packet_size)
    if not data:
      return
    filepath_len, filename_len, filepath,filename, filesize = struct.unpack(head_packet_format,data)
    filepath = os.path.join(nagios_perfdata, filepath[:filepath_len])
    filename = filename[:filename_len]
    self.logger.debug("update file: %s" % filepath + '/' + filename)
    try:
      if not os.path.exists(filepath):
        os.makedirs(filepath)
    except OSError:
      pass
    self.fd = open(os.path.join(filepath,filename), 'w')
    #self.fd = open(filename,'w')
    if filesize > self.chunk_size:
      times = filesize / self.chunk_size
      first_part_size = times * self.chunk_size
      second_part_size = filesize % self.chunk_size
      while 1:
        try:
          data = self.recv(self.chunk_size)
          #self.logger.debug("handle_read()->%s size.",len(data))
        except socket.error,e:
          if e.args[0] == errno.EWOULDBLOCK:
            print "EWOULDBLOCK"
            time.sleep(1)
          else:
            #self.logger.debug("Error happend while receive data: %s" % e)
            break
        else:
          self.fd.write(data)
          self.fd.flush()
          self.writen_size += len(data)
          if self.writen_size == first_part_size:
            break
      #receive the packet at last
      while 1:
        try:
          data = self.recv(second_part_size)
          #self.logger.debug("handle_read()->%s size.",len(data))
        except socket.error,e:
          if e.args[0] == errno.EWOULDBLOCK:
            print "EWOULDBLOCK"
            time.sleep(1)
          else:
            #self.logger.debug("Error happend while receive data: %s" % e)
            break
        else:
          self.fd.write(data)
          self.fd.flush()
          self.writen_size += len(data)
          if len(data) == second_part_size:
            break
    elif filesize <= self.chunk_size:
      while 1:
        try:
          data = self.recv(filesize)
          #self.logger.debug("handle_read()->%s size.",len(data))
        except socket.error,e:
          if e.args[0] == errno.EWOULDBLOCK:
            print "EWOULDBLOCK"
            time.sleep(1)
          else:
            #self.logger.debug("Error happend while receive data: %s" % e)
            break
        else:
          self.fd.write(data)
          self.fd.flush()
          self.writen_size += len(data)
          if len(data) == filesize:
            break
    self.logger.debug("File size: %s" % self.writen_size)
class SyncServer(asyncore.dispatcher):
  def __init__(self,host,port):
    asyncore.dispatcher.__init__(self)
    self.debug = True
    self.logger = logging.getLogger(self.__class__.__name__)
    self.create_socket(socket.AF_INET,socket.SOCK_STREAM)
    self.set_reuse_addr()
    self.bind((host,port))
    self.listen(2000)
  def handle_accept(self):
    client_socket = self.accept()
    if client_socket is None:
      pass
    else:
      sock, addr = client_socket
      #self.logger.debug("Incoming connection from %s" % repr(addr))
      handler = RequestHandler(sock=sock)
class RunServer(threading.Thread):
  def __init__(self):
    super(RunServer,self).__init__()
    self.daemon = False
  def run(self):
    server = SyncServer('',9999)
    asyncore.loop(use_poll=True)
def StartServer():
  logging.basicConfig(level=logging.DEBUG,
            format='%(name)s: %(message)s',
            )
  RunServer().start()
  #MakeGraph().start()
if __name__ == '__main__':
  StartServer()

客户端：

# monitor path with inotify(python module), and send them to remote server.#
# use sendfile(2) instead of send function in socket, if we have python-sendfile installed.#
import socket
import time
import os
import sys
import struct
import threading
import Queue
try:
   import pyinotify
except (ImportError, ImportWarnning):
   print "Hope this information can help you:"
   print "Can not find pyinotify module in sys path, just run [apt-get install python-pyinotify] in ubuntu."
   sys.exit(1)
try:
   from sendfile import sendfile
except (ImportError,ImportWarnning):
   pass
filetype_filter = [".rrd",".xml"]
def check_filetype(pathname):
   for suffix_name in filetype_filter:
     if pathname[-4:] == suffix_name:
       return True
   try:
     end_string = pathname.rsplit('.')[-1:][0]
     end_int = int(end_string)
   except:
     pass
   else:
     # means pathname endwith digit
     return False
class sync_file(threading.Thread):
   def __init__(self, addr, events_queue):
     super(sync_file,self).__init__()
     self.daemon = False
     self.queue = events_queue
     self.addr = addr
     self.chunk_size = 1024
   def run(self):
     while 1:
       event = self.queue.get()
       if check_filetype(event.pathname):
         print time.asctime(),event.maskname, event.pathname
         filepath = event.path.split('/')[-1:][0]
         filename = event.name
         filesize = os.stat(os.path.join(event.path, filename)).st_size
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         filepath_len = len(filepath)
         filename_len = len(filename)
         sock.connect(self.addr)
         offset = 0
         data = struct.pack("!LL128s128sL",filepath_len, filename_len, filepath,filename,filesize)
         fd = open(event.pathname,'rb')
         sock.sendall(data)
         if "sendfile" in sys.modules:
           # print "use sendfile(2)"
           while 1:
             sent = sendfile(sock.fileno(), fd.fileno(), offset, self.chunk_size)
             if sent == 0:
               break
             offset += sent
         else:
           # print "use original send function"
           while 1:
             data = fd.read(self.chunk_size)
             if not data: break
             sock.send(data)
         sock.close()
         fd.close()
class EventHandler(pyinotify.ProcessEvent):
   def __init__(self, events_queue):
     super(EventHandler,self).__init__()
     self.events_queue = events_queue
   def my_init(self):
     pass
   def process_IN_CLOSE_WRITE(self,event):
     self.events_queue.put(event)
   def process_IN_MOVED_TO(self,event):
     self.events_queue.put(event)
def start_notify(path, mask, sync_server):
   events_queue = Queue.Queue()
   sync_thread_pool = list()
   for i in range(500):
     sync_thread_pool.append(sync_file(sync_server, events_queue))
   for i in sync_thread_pool:
     i.start()
   wm = pyinotify.WatchManager()
   notifier = pyinotify.Notifier(wm,EventHandler(events_queue))
   wdd = wm.add_watch(path,mask,rec=True)
   notifier.loop()
def do_notify():
   perfdata_path = '/var/lib/pnp4nagios/perfdata'
   mask = pyinotify.IN_CLOSE_WRITE|pyinotify.IN_MOVED_TO
   sync_server = ('127.0.0.1',9999)
   start_notify(perfdata_path,mask,sync_server)
if __name__ == '__main__':
   do_notify()

python监视线程池

#!/usr/bin/python
import threading
import time
class Monitor(threading.Thread):
  def __init__(self, *args,**kwargs):
    super(Monitor,self).__init__()
    self.daemon = False
    self.args = args
    self.kwargs = kwargs
    self.pool_list = []
  def run(self):
    print self.args
    print self.kwargs
    for name,value in self.kwargs.items():
      obj = value[0]
      temp = {}
      temp[name] = obj
      self.pool_list.append(temp)
    while 1:
      print self.pool_list
      for name,value in self.kwargs.items():
        obj = value[0]
        parameters = value[1:]
        died_threads = self.cal_died_thread(self.pool_list,name)
        print "died_threads", died_threads
        if died_threads >0:
          for i in range(died_threads):
            print "start %s thread..." % name
            t = obj[0].__class__(*parameters)
            t.start()
            self.add_to_pool_list(t,name)
        else:
          break
      time.sleep(0.5)
  def cal_died_thread(self,pool_list,name):
    i = 0
    for item in self.pool_list:
      for k,v in item.items():
        if name == k:
          lists = v
    for t in lists:
      if not t.isAlive():
        self.remove_from_pool_list(t)
        i +=1
    return i
  def add_to_pool_list(self,obj,name):
    for item in self.pool_list:
      for k,v in item.items():
        if name == k:
          v.append(obj)
  def remove_from_pool_list(self, obj):
    for item in self.pool_list:
      for k,v in item.items():
        try:
          v.remove(obj)
        except:
          pass
        else:
          return

使用方法：

rrds_queue = Queue.Queue()
  make_rrds_pool = []
  for i in range(5):
    make_rrds_pool.append(MakeRrds(rrds_queue))
  for i in make_rrds_pool:
    i.start()
  make_graph_pool = []
  for i in range(5):
    make_graph_pool.append(MakeGraph(rrds_queue))
  for i in make_graph_pool:
    i.start()
  monitor = Monitor(make_rrds_pool=(make_rrds_pool, rrds_queue), \
           make_graph_pool=(make_graph_pool, rrds_queue))
  monitor.start()

解析：

1. 接受字典参数，value为一个元组，第一个元素是线程池，后面的都是参数。
2. 每0.5秒监视线程池中的线程数量，如果线程死掉了，记录死掉线程的数目，再启动同样数量的线程。
3. 如果没有线程死去，则什么也不做。

从外部调用Django模块

import os
import sys
sys.path.insert(0,'/data/cloud_manage')
from django.core.management import setup_environ
import settings
setup_environ(settings)
from common.monitor import Monitor
from django.db import connection, transaction

前提就是，要新建一个django的project，这里我们新建了一个cloud_manage.
这样不仅可以调用django自身的模块，还能调用project本身的东西。

希望本文所述对大家的Python程序设计有所帮助。