使用 Autobahn Testsuite 进行 Websocket 压力测试

Question

我尝试对我的 websocket 服务器做一些压力测试。在客户端，我运行以下脚本 from this site :

 import time, sys
 from twisted.internet import defer, reactor
 from twisted.internet.defer import Deferred, returnValue, inlineCallbacks

 from autobahn.twisted.websocket import connectWS, \
                                   WebSocketClientFactory, \
                                   WebSocketClientProtocol


 class MassConnectProtocol(WebSocketClientProtocol):

  didHandshake = False

  def onOpen(self):

    print("websocket connection opened")
    self.factory.test.onConnected()
    self.factory.test.protos.append(self)
    self.didHandshake = True


class MassConnectFactory(WebSocketClientFactory):

 protocol = MassConnectProtocol

  def clientConnectionFailed(self, connector, reason):
   if self.test.onFailed():
     reactor.callLater(float(self.retrydelay)/1000., connector.connect)

  def clientConnectionLost(self, connector, reason):
   if self.test.onLost():
     reactor.callLater(float(self.retrydelay)/1000., connector.connect)


class MassConnect:

  def __init__(self, name, uri, connections, batchsize, batchdelay, retrydelay):
   print('MassConnect init')
   self.name = name
   self.uri = uri
   self.batchsize = batchsize
   self.batchdelay = batchdelay
   self.retrydelay = retrydelay
   self.failed = 0
   self.lost = 0
   self.targetCnt = connections
   self.currentCnt = 0
   self.actual = 0
   self.protos = []

 def run(self):
   print('MassConnect runned')
   self.d = Deferred()
   self.started = time.clock()
   self.connectBunch()
   return self.d

 def onFailed(self):
   self.failed += 1
   sys.stdout.write("!")
   return True

 def onLost(self):
   self.lost += 1
   #sys.stdout.write("*")
   return False
   return True

 def onConnected(self):
   print("onconnected")
   self.actual += 1
   if self.actual % self.batchsize == 0:
     sys.stdout.write(".")
   if self.actual == self.targetCnt:
     self.ended = time.clock()
     duration = self.ended - self.started
     print " connected %d clients to %s at %s in %s seconds (retries %d = failed %d + lost %d)" % (self.currentCnt, self.name, self.uri, duration, self.failed + self.lost, self.failed, self.lost)
     result = {'name': self.name,
               'uri': self.uri,
               'connections': self.targetCnt,
               'retries': self.failed + self.lost,
               'lost': self.lost,
               'failed': self.failed,
               'duration': duration}
     for p in self.protos:
        p.sendClose()
     #self.d.callback(result)

  def connectBunch(self):
    if self.currentCnt + self.batchsize < self.targetCnt:
      c = self.batchsize
      redo = True
    else:
      c = self.targetCnt - self.currentCnt
      redo = False
    for i in xrange(0, c):
      factory = MassConnectFactory(self.uri)
      factory.test = self
      factory.retrydelay = self.retrydelay
      connectWS(factory)
      self.currentCnt += 1
    if redo:
      reactor.callLater(float(self.batchdelay)/1000., self.connectBunch)


class MassConnectTest:
  def __init__(self, spec):
   self.spec = spec
   print('MassConnetest init')
 @inlineCallbacks
 def run(self):
  print self.spec
  res = []
  for s in self.spec['servers']:
     print s['uri']
     t = MassConnect(s['name'],
                     s['uri'],
                     self.spec['options']['connections'],
                     self.spec['options']['batchsize'],
                     self.spec['options']['batchdelay'],
                     self.spec['options']['retrydelay'])
     r = yield t.run()
     res.append(r)
  returnValue(res)


def startClient(spec, debug = False):
  test = MassConnectTest(spec)
  d = test.run()
  return d

if __name__ == '__main__':

  spec = {}
  spec['servers'] = [{'name': 'test', 'uri':"ws://127.0.0.1:8080"} ]
  spec['options'] ={'connections': 1000,'batchsize': 500, 'batchdelay': 1000, 'retrydelay': 200 }
  startClient(spec,False)

但是在运行执行此脚本后，服务器端没有建立任何连接。服务器似乎配置正确，因为当我使用不同的客户端（例如网络浏览器）连接到我的服务器时，它工作正常并且建立了 websocket 连接。我还检查了网络嗅探器，似乎脚本没有产生任何 websocket 连接。我在这个脚本中做错了什么？

Answer 1

您使用的 massconnect.py 脚本应该从高速公路测试套件的另一部分调用，例如 wstest 命令：

$ echo '{"servers": [{"name": "test", "uri":"ws://127.0.0.1:8080"} ], "options": {"connections": 1000,"batchsize": 500, "batchdelay": 1000, "retrydelay": 200 }}' > spec.json
$ wstest -m massconnect --spec spec.json

如果你想直接复制massconnect，我认为它缺少启动Twisted延迟任务的命令：

if __name__ == '__main__':
  spec = {}
  spec['servers'] = [{'name': 'test', 'uri':"ws://127.0.0.1:8080"} ]
  spec['options'] ={'connections': 1000,'batchsize': 500, 'batchdelay': 1000, 'retrydelay': 200 }
  startClient(spec,False)
  reactor.run() # <-- add this

并检查您的 Python 缩进，其中一些缩进在粘贴到此处时已损坏，或者原始代码在某些 class 和函数定义中的缩进不正确。

使用 Autobahn Testsuite 进行 Websocket 压力测试

Websocket stress test with Autobahn Testsuite

websocket

python-2.7

autobahn