快速将通过套接字传输的字符串数据转化为Python中的对象

Question

我现在有一个Python应用程序，通过TCP/IP套接字接收以换行符结束的ASCII字符串。这些字符串的传输速度很快，我需要尽可能快地解析它们。目前，这些字符串是以CSV格式传输的，如果数据传输速度太快，我的Python应用就会跟不上输入的数据速度（这可能也不算意外）。

这些字符串大概长这样：

chan,2007-07-13T23:24:40.143,0,0188878425-079,0,0,True,S-4001,UNSIGNED_INT,name1,module1,...

我有一个对应的对象来解析这些字符串，并将所有数据存储到一个对象中。目前这个对象大概是这样的：

class ChanVal(object):
    def __init__(self, csvString=None,**kwargs):

        if csvString is not None:
            self.parseFromCsv(csvString)

        for key in kwargs:
                setattr(self,key,kwargs[key])

    def parseFromCsv(self, csvString):

        lst = csvString.split(',')

        self.eventTime=lst[1]
        self.eventTimeExact=long(lst[2])
        self.other_clock=lst[3]
        ...

为了从套接字读取数据，我使用了一个基本的“socket.socket(socket.AF_INET,socket.SOCK_STREAM)”（我的应用是服务器套接字），然后我使用“select”模块中的“select.poll()”对象，不断地用它的“poll(...)"方法来检查套接字是否有新输入。

我对发送数据的过程有一些控制（也就是说，我可以让发送方改变格式），但如果我们能加快ASCII处理的速度，就不需要使用固定宽度或二进制格式的数据，那就太方便了。

到目前为止，我尝试过的一些方法并没有太大改善：

1. 使用字符串的“split”方法，然后直接索引结果列表（见上文），但“split”似乎真的很慢。
2. 使用“csv”模块中的“reader”对象来解析字符串。
3. 将发送的字符串改成我可以直接用“eval”实例化对象的格式（例如，发送类似“ChanVal(eventTime='2007-07-13T23:24:40.143',eventTimeExact=0,...)”的内容）。

我想避免使用固定宽度或二进制格式，虽然我意识到这些可能最终会更快。

总的来说，我希望能得到一些建议，比如更好的轮询套接字的方法，更好的数据格式/解析方式（希望我们能继续使用ASCII），或者你能想到的其他任何建议。

谢谢！

Answer 1

你需要对你的代码进行性能分析，找出时间都花在哪里了。

这并不一定要用Python自带的性能分析工具。

比如，你可以尝试用不同的方法解析同一个CSV字符串一百万次。选择最快的方法，然后把这个时间除以一百万，这样你就知道解析一个字符串大概需要多少CPU时间了。

试着把程序分成几个部分，看看每个部分到底需要多少资源。

那些每处理一行输入就需要最多CPU的部分就是你的瓶颈。

在我的电脑上，下面的程序输出了这个结果：

ChanVal0 took 0.210402965546 seconds
ChanVal1 took 0.350302934647 seconds
ChanVal2 took 0.558166980743 seconds
ChanVal3 took 0.691503047943 seconds

所以你可以看到，大约一半的时间都花在了parseFromCsv上。不过，提取值并存储到类中也花了不少时间。

如果这个类不马上用到，可能存储原始数据会更快，然后在需要的时候再用属性来解析CSV字符串。

from time import time
import re

class ChanVal0(object):
    def __init__(self, csvString=None,**kwargs):
        self.csvString=csvString
        for key in kwargs:
            setattr(self,key,kwargs[key])

class ChanVal1(object):
    def __init__(self, csvString=None,**kwargs):
        if csvString is not None:
            self.parseFromCsv(csvString)
        for key in kwargs:
                setattr(self,key,kwargs[key])

    def parseFromCsv(self, csvString):
        self.lst = csvString.split(',')

class ChanVal2(object):
    def __init__(self, csvString=None,**kwargs):
        if csvString is not None:
            self.parseFromCsv(csvString)
        for key in kwargs:
                setattr(self,key,kwargs[key])

    def parseFromCsv(self, csvString):
        lst = csvString.split(',')
        self.eventTime=lst[1]
        self.eventTimeExact=long(lst[2])
        self.other_clock=lst[3]


class ChanVal3(object):
    splitter=re.compile("[^,]*,(?P<eventTime>[^,]*),(?P<eventTimeExact>[^,]*),(?P<other_clock>[^,]*)")
    def __init__(self, csvString=None,**kwargs):
        if csvString is not None:
            self.parseFromCsv(csvString)
        self.__dict__.update(kwargs)

    def parseFromCsv(self, csvString):
        self.__dict__.update(self.splitter.match(csvString).groupdict())


s="chan,2007-07-13T23:24:40.143,0,0188878425-079,0,0,True,S-4001,UNSIGNED_INT,name1,module1"
RUNS=100000

for cls in ChanVal0, ChanVal1, ChanVal2, ChanVal3:
    start_time = time()
    for i in xrange(RUNS):
        cls(s)
    print "%s took %s seconds"%(cls.__name__, time()-start_time) 

Answer 2

你不能让Python本身变得更快，但你可以让你的Python应用程序运行得更快。

原则一：减少工作量。

你不能在整体上减少输入解析的工作，但你可以在读取数据和处理其他事情的过程中减少输入解析的工作。

通常，你可以这样做。

把你的应用程序分成几个独立的步骤。

1. 读取数据，分解成不同的字段，创建一个命名元组，然后用类似pickle的方式把这个元组写入一个管道。

2. 从管道中读取数据（使用pickle），构建命名元组，进行一些处理，然后写入另一个管道。

3. 从管道中读取数据，进行一些处理，然后写入文件或其他地方。

这三个过程通过操作系统的管道连接在一起，可以同时运行。这意味着第一个过程在读取数据并创建元组，而第二个过程在使用这些元组进行计算，第三个过程则在进行计算并写入文件。

这种管道方式最大化了你的CPU的使用效率，而且不需要太多复杂的技巧。

在Linux中，读取和写入管道是很简单的，因为系统会确保在创建管道时，sys.stdin和sys.stdout会是管道。

在做其他事情之前，先把你的程序分成几个管道阶段。

proc1.py

import cPickle
from collections import namedtuple

ChanVal= namedtuple( 'ChanVal', ['eventTime','eventTimeExact', 'other_clock', ... ] )
for line socket:
    c= ChanVal( **line.split(',') )
    cPickle.dump( sys.stdout )

proc2.py

import cPickle
from collections import namedtuple
ChanVal= namedtuple( 'ChanVal', ['eventTime','eventTimeExact', 'other_clock', ... ] )
while True:
    item = cPickle.load( sys.stdin )
    # processing
    cPickle.dump( sys.stdout )

通过管道处理命名元组的这个想法是非常可扩展的。

python proc1.py | python proc2.py