python socket server带有粘包、分包处理的服务端写法

5250阅读　0评论2016-03-22　lolizeppelin
分类：Python/Ruby

用epoll写的,换select写法也差不多


点击(此处)折叠或打开

  1. # -*- coding: UTF-8 -*-
  2. import re
  3. import socket
  4. import select
  5. import logging
  6. import struct
  7. import time

  9. # 里面存放自定义常量
  10. import _static

  12. __author__ = 'loliz_000'


  15. def decode_socket_data(data_buffer):
  16.     """
  17.     命令包解析函数,所有包都以WPROJ_PACKAGER_PACK_HEAD + 包命令类型为开头 包内存在string,则第三部分为string的长度
  18.     @param data_buffer:
  19.     @return 返回none,解析失败
  20.     @return 返回结构 (剩余半包, [(包1命令码,包1结果码,包1内容), (包2命令码,包2结果码,包2内容)])
  21.     """
  22.     data_list = []
  23.     buffer_size = len(data_buffer)
  24.     if buffer_size < 12:
  25.         return data_buffer, data_list
  26.     try:
  27.         # 这里是我自定义包结构
  28.         wproj_head_code, package_command_type, code, package_len = struct.unpack('>HHII', data_buffer[0:12])
  29.         # 解包,判断包头码
  30.         if wproj_head_code != _static.WPROJ_PACKAGER_PACK_HEAD:
  31.             return None
  32.     except struct.error:
  33.         return None
  34.     # 包长度过长
  35.     if package_len >= _static.WPROJ_PACKAGER_MAX_PACK_LEN:
  36.         logging.warning('decode_socket_data find data over max size')
  37.         return None
  38.     # 没有包体
  39.     if buffer_size == 12:
  40.         if package_len > 0:
  41.             return None
  42.         data_list.append((package_command_type, code, None))
  43.         return None, data_list
  44.     # 所需包长度 > 实际包长, 这是一个半包
  45.     if package_len > len(data_buffer[12:]):
  46.         return data_buffer, data_list
  47.     # 包长度正好
  48.     elif package_len == len(data_buffer[12:]):
  49.         try:
  50.             data = struct.unpack('>%ds' % package_len, data_buffer[12:])
  51.         except struct.error:
  52.             return None
  53.         data_list.append((package_command_type, code, data))
  54.         return None, data_list
  55.     # 所需包长度 < 实际包长 有粘包,递归处理
  56.     else:
  57.         try:
  58.             data = struct.unpack('>%ds' % package_len, data_buffer[12:12+package_len])
  59.         except struct.error:
  60.             return None
  61.         data_list.append((package_command_type, code, data))
  62.         decode_socket_res = decode_socket_data(data_buffer[12+package_len:])
  63.         if not decode_socket_res:
  64.             return None
  65.         return decode_socket_res[0], data_list.extend(decode_socket_res[1])



  69. class AgentServer:
  70.     """
  71.     打包端守护进程
  72.     """
  73.     LOG_TYPE_RE = re.compile('^((DEBUG)|(INFO)|(WARNING)|(ERROR)|(CRITICAL))+?$', re.IGNORECASE)

  75.     def __init__(self):
  76.         self.bind_port = 2000
  77.         self.bind_sock = None
  78.         self.epoll = None
  79.         self.bind_lock = None
  80.         self.queue = None
  81.         self.socket_pool_dict = {}

  83.     def bind_sock_start(self):
  84.         logging.info('PackageAgentServer bind socket start')
  85.         bind_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  86.         #bind_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  87.         try:
  88.             socket.setdefaulttimeout(10)
  89.             bind_sock.bind(('0.0.0.0', self.bind_port))
  90.             bind_sock.listen(100)
  91.             self.bind_sock = bind_sock
  92.             logging.warning('PackageAgentServer bind socket success')
  93.         except socket.error:
  94.             logging.warning('PackageAgentServer bind socket get socket error')
  95.             return False
  96.         except OSError:
  97.             logging.warning('PackageAgentServer bind socket get os error')
  98.             return False
  99.         except IOError:
  100.             logging.warning('PackageAgentServer bind socket get io error')
  101.             return False
  102.         logging.info('PackageAgentServer bind socket end')
  103.         self.epoll = select.epoll()
  104.         self.epoll.register(self.bind_sock.fileno(), select.EPOLLIN)
  105.         return True


  108.     def add_to_socket_pool(self, sock):
  109.         self.bind_lock.acquire()
  110.         # socket 连接池 添加
  111.         self.epoll.register(sock.fileno(), select.EPOLLIN | select.EPOLLET)
  112.         self.socket_pool_dict[sock.fileno] = (sock, None)
  113.         self.bind_lock.release()

  115.     def del_from_socket_pool(self, sock):
  116.         if isinstance(sock, socket.socket):
  117.             fd = sock.fileno()
  118.         else:
  119.             fd = sock
  120.         self.bind_lock.acquire()
  121.         try:
  122.             # socket 连接池移除
  123.             self.socket_pool_dict.pop(fd)
  124.             self.epoll.unregister(fd, select.EPOLLIN | select.EPOLLET)
  125.         except KeyError:
  126.             logging.warning('del_from_socket_pool pop sock more then once')
  127.         self.bind_lock.release()

  129.     def connection_loop(self):
  130.         """
  131.         获取新tcp-ip连接,另启一个线程运行
  132.         """
  133.         error_count = 0
  134.         logging.info('scoket connection_loop start')
  135.         while 1: # 这个循不在主线程里运行,fork后不需要处理
  136.             # 退出标记为1结束循环
  137.             if self.single_out:
  138.                 break
  139.             if error_count > 500:
  140.                 logging.warning('connection_loop get too much error')
  141.                 self.single_out = 1
  142.                 continue
  143.             for fd, events in self.epoll.poll():
  144.                 if fd == self.bind_sock.fileno():
  145.                     client_sock, client_ipaddr = self.bind_sock.accept()
  146.                     self.add_to_socket_pool(client_sock)
  147.                 elif select.EPOLLIN & events:
  148.                     cur_sock, half_package = self.socket_pool_dict[fd]
  149.                     data = cur_sock.recv(10000)
  150.                     # 有半包数据
  151.                     if half_package:
  152.                         data = half_package + data
  153.                     # 调用解析数据包
  154.                     decode_data = decode_socket_data(data)
  155.                     # 包解析出错,返回None
  156.                     if not decode_data:
  157.                         # 自己写个关闭函数
  158.                         close_ret = close_sock(cur_sock, _static.DECODE_PACKAGE_FAILURE, 'package decode failure')
  159.                         if close_ret['ret'] > 0:
  160.                             logging.error('close socket error at DECODE_PACKAGE, error %s' % close_ret['msg'])
  161.                         self.del_from_socket_pool(fd)
  162.                         continue
  163.                     # 包解析正确, 半包传入socket_pool_dict,供下次数据处理用
  164.                     self.socket_pool_dict[fd][1], data_list = decode_data
  165.                     # 已经处理好的数据列表塞入队列, 这里可以做点优化, 直接在解析包的函数中就塞入队列
  166.                     for _data in data_list:
  167.                             self.queue.put((cur_sock, _data))
  168.                 elif select.EPOLLHUP & events:
  169.                     self.del_from_socket_pool(fd)
  170.                 else:
  171.                     logging.warning('unkonw event')


  174.     def work_loop(self):
  175.         while 1:
  176.             # 退出标记为1,结束循环
  177.             if self.single_out:
  178.                 break
  179.             if self.queue.empty():
  180.                 time.sleep(0.01)
  181.             else:
  182.                 queue_data = self.queue.get()
  183.                 cur_sock = queue_data[0]
  184.                 package_command_type, code, data = queue_data[1]
  185.                 # do what you want do

  187.     def start(self):
  188.         # 绑定端口
  189.         if not self.bind_sock_start():
  190.             raise
  191.         import threading
  192.         # 启动数据接收线程
  193.         data_thread = threading.Thread(target=self.connection_loop)
  194.         data_thread.setDaemon(True)
  195.         data_thread.start()
  196.         # 启动数据处理线程
  197.         data_thread = threading.Thread(target=self.work_loop)
  198.         data_thread.setDaemon(True)
  199.         data_thread.start()


上一篇：windows非阻塞读取管道内容
下一篇：7z命令行使用