3. WebSocket并发实战

3.1 整理并发需求

为了便于理解，咱们的并发需求及指标，还是参照小鱼的这篇《接口并发性能测试开发之：从测试方案设计、测试策略、指标分析到代码编写，这一篇全搞定。》。

需求内容：

中午和晚上是订餐的高峰期，除了订单的并发量很大以外，对老板的并发量也很大，而老板的订单通知则是通过WebSocket来发送，获取订单通知之前需要建立WebSocket连接，所以需要通过并发来测试客户端对服务器的WebSocket并发连接成功率。

3.2 提取性能指标

性能指标提取

这个场景，仅仅测试WebSocket并发连接成功率即可。

以客户端向服务器发送100个并发的WebSocket请求为例，需要提取并发指标如下：

    · 连接数

    · 成功连接数

    · 成功连接率

    · 性能指标分析

连接数就是自定义的并发，因为请求简单而且执行比较快，所以直接用100个线程并发，不需要循环次数。

连接成功数就是获取WebSocket的响应数，

连接成功率只需要把成功连接数/总连接数。

3.3 代码编写

还是老规矩，直接上代码，在代码中分析

# -*- coding:utf-8 -*-
# @Time   : 2022-05-20
# @Author : carl_DJ

import threading
import websocket

#初始化几个全局变量
#定义并发数100
THREAD_NUM = 100
#自定义全局变量每个线程需要循环的数量,1
ONE_WORKER_NUM = 1
#设定最开始的成功连接数
success_count = 0

#编写WebSocket连接请求
def websockettest():
	
	global success_count
	#定义url
	url = "ws://www.xxx.com/serv/push?chann = delivery"
	
    #使用create_connection方法发送url连接
	ws = websocket.create_connetcion(url)
	
	#连接完成之后通过send()方法发送请求内容
	ws.send("{'type':'heartbeat','username':'13600001111','message':'ok'}")
	
	#将返回的信息赋值给new_masc
	msg = ws.recv()
	#连接成功就+1，获取成功次数
	success_count = success_count + 1

'''嵌套指定循环次数的order()函数'''

def  working():
	global ONE_WORKER_NUM
	for i in range(0,ONE_WORKER_NUM):
		websockettest()

 '''自定义main（）函数，来执行线程'''
def t():
    #自定义一个空的数组，用来存放线程组
	Threads = []
	#设置循环次数
	for i in range(THREAD_NUM):
		#将working()函数存放到线程中
        t = threading.Thread(target=working,name="T"+ str(i))
         #设定守护线程
        t.setDaemon(True)
        threads.append(t)
     
    #启动循环次数
	for  t in Threads():
		t.start()
    ##设置阻塞线程
	for t in Threads:
        t.join()
    print ("main thread end")

if __name__ == "__main__":
	t()
	# 结果写入文件
	result = open("E:\Progect\resut.txt","a")
	#总连接次数
	result.write("发起连接数:" + str(THREAD_NUM) + "\n" )
	#成功连接次数
	result.write("成功连接数:" + str(success_count) + "\n" )
	#成功链接率
	result.write("成功链接率:" + str(success_count/THREAD_NUM *100) + "%" + "\n" )
	result.close()