作者 | 冬岛  阿里云容器平台工程师
 
 
 
导读：虽然说 Knative 默认就支持 WebSocket 和 gRPC，但在使用中会发现，有时想要把自己的 WebSocket 或 gRPC 部署到 Knative 中，还是存在各种不顺利。虽然最后排查发现，大多是自己的程序问题或是配置错误导致的。本文分别给出了一个 WebSocket 和 gRPC 的例子，当需要在生产或者测试环境部署相关服务时，可以使用本文给出的示例进行 Knative 服务的测试。
 
 
WebSocket
 
如果自己手动的配置 Istio Gateway 支持 WebSocket 就需要开启 websocketUpgrade 功能。但使用 Knative Serving 部署其实就自带了这个能力。本示例的完整代码放在 https://github.com/knative-sample/websocket-chat ，这是一个基于 WebSocket 实现的群聊的例子。
 
使用浏览器连接到部署的服务中就可以看到一个接收信息的窗口和发送信息的窗口。当你发出一条信息以后所有连接进来的用户都能收到你的消息。所以你可以使用两个浏览器窗口分别连接到服务中，一个窗口发送消息一个窗口接收消息，以此来验证 WebSocket 服务是否正常。
 
本示例是在 gorilla/websocket  基础之上进行了一些优化：
 
代码中添加了 vendor 依赖，你下载下来就可以直接使用
添加了 Dockerfile 和 Makefile 可以直接编译二进制和制作镜像
添加了 Knative Sevice 的 yaml 文件(service.yaml)，你可以直接提交到 Knative 集群中使用
也可以直接使用编译好的镜像 registry.cn-hangzhou.aliyuncs.com/knative-sample/websocket-chat:2019-10-15
Knative Service 配置：
 
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: websocket-chat
spec:
  template:
    spec:
      containers:
        - image: registry.cn-hangzhou.aliyuncs.com/knative-sample/websocket-chat:2019-10-15
          ports:
            - name: http1
              containerPort: 8080
 
代码 clone 下来以后执行 kubectl apply -f service.yaml 把服务部署到 Knative 中，然后直接访问服务地址即可使用。
 
查看 ksvc 列表，并获取访问域名。
 
└─# kubectl get ksvc
NAME             URL                                                    LATESTCREATED          LATESTREADY            READY   REASON
websocket-chat   http://websocket-chat.default.serverless.kuberun.com   websocket-chat-7ghc9   websocket-chat-7ghc9   True
 
现在使用浏览器打开 http://websocket-chat.default.serverless.kuberun.com 即可看到群聊窗口。
 
 
打开两个窗口，在其中一个窗口发送一条消息，另外一个窗口通过 WebSocket 也收到了这条消息。
 
gRPC
 
gRPC 不能通过浏览器直接访问，需要通过 client 端和 server 端进行交互。本示例的完整代码放在 https://github.com/knative-sample/grpc-ping-go ，本示例会给一个可以直接使用的镜像，测试 gRPC 服务。
 
Knative Service 配置：
 
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: grpc-ping
spec:
  template:
    spec:
      containers:
      - image: registry.cn-hangzhou.aliyuncs.com/knative-sample/grpc-ping-go:2019-10-15
        ports:
          - name: h2c
            containerPort: 8080
 
代码 clone 下来以后执行 kubectl apply -f service.yaml 把服务部署到 Knative 中。
 
获取 ksvc 列表和访问域名：
 
└─# kubectl get ksvc
NAME             URL                                                    LATESTCREATED          LATESTREADY            READY     REASON
grpc-ping        http://grpc-ping.default.serverless.kuberun.com        grpc-ping-p2tft                               Unknown   RevisionMissing
websocket-chat   http://websocket-chat.default.serverless.kuberun.com   websocket-chat-6hgld   websocket-chat-6hgld   True
 
现在我们已经知道 gRPC  server 的地址是 grpc-ping.default.serverless.kuberun.com，端口是 80，那么我们可以发起测试请求：
 
└─# docker run --rm registry.cn-hangzhou.aliyuncs.com/knative-sample/grpc-ping-go:2019-10-15 /client -server_addr="grpc-ping.default.serverless.kuberun.com:80" -insecure
2019/10/16 11:35:07 Ping got hello - pong
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854794231 +0800 CST m=+73.061909052
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854827273 +0800 CST m=+73.061942072
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854835802 +0800 CST m=+73.061950606
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854842843 +0800 CST m=+73.061957643
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854849211 +0800 CST m=+73.061964012
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854855249 +0800 CST m=+73.061970049
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854861659 +0800 CST m=+73.061976460
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854875071 +0800 CST m=+73.061989873
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854905416 +0800 CST m=+73.062020221
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.85491183 +0800 CST m=+73.062026630
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.85492533 +0800 CST m=+73.062040133
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854932285 +0800 CST m=+73.062047083
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854946977 +0800 CST m=+73.062061782
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854953311 +0800 CST m=+73.062068112
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854966639 +0800 CST m=+73.062081440
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854973939 +0800 CST m=+73.062088739
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854985463 +0800 CST m=+73.062100268
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854993275 +0800 CST m=+73.062108073
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.854999812 +0800 CST m=+73.062114613
2019/10/16 11:35:07 Got pong 2019-10-16 11:35:07.855012676 +0800 CST m=+73.062127479
 
小结
 
本文通过两个例子分别展示了 WebSocket 和 gRPC 的部署方法：
 
WebSocket 示例通过一个 chat 的方式展示发送和接受消息
gRPC 通过启动一个 client 的方式展示 gRPC 远程调用的过程
原文链接
 本文为云栖社区原创内容，未经允许不得转载。

 
 
本文转载：https://github.com/alberliu/gim
 
 
简要介绍
 
gim是一个即时通讯服务器，代码全部使用golang完成。主要功能
 1.支持tcp，websocket接入
 2.离线消息同步
 3.多业务接入
 4.单用户多设备同时在线
 5.单聊，群聊，以及超大群聊天场景
 6.支持服务水平扩展
 
使用技术：
 
数据库：Mysql+Redis
 通讯框架：Grpc
 长连接通讯协议：Protocol Buffers
 日志框架：Zap
 
安装部署
 
1.首先安装MySQL，Redis
 2.创建数据库gim，执行sql/create_table.sql，完成初始化表的创建（数据库包含提供测试的一些初始数据）
 3.修改config下配置文件，使之和你本地配置一致
 4.分别切换到cmd的tcp_conn,ws_conn,logic目录下，执行go run main.go,启动TCP连接层服务器,WebSocket连接层服务器,逻辑层服务器
 
迅速跑通本地测试
 
1.在test目录下，tcp_conn或者ws_conn目录下，执行go run main,启动测试脚本
 2.根据提示,依次填入app_id,user_id,device_id,sync_sequence(中间空格空开)，进行长连接登录；数据库device表中已经初始化了一些设备信息，用作测试
 3.执行api/logic/logic_client_ext_test.go下的TestLogicExtServer_SendMessage函数，发送消息
 
业务服务器如何接入
 
1.首先生成私钥和公钥
 2.在app表里根据你的私钥添加一条app记录
 3.将app_id和公钥保存到业务服务器
 4.将用户通过LogicClientExtServer.AddUser接口添加到IM服务器
 5.通过LogicClientExtServer.RegisterDevice接口注册设备，获取设备id(device_id)
 6.将app_id，user_id,device_id用公钥通过公钥加密，生成token,相应库的代码在pkg/util/aes.go
 7.接下来使用这个token，app就可以和IM服务器交互
 
rpc接口简介
 
项目所有的proto协议在gim/public/proto/目录下
 1.tcp.proto
 长连接通讯协议
 2.logic_client.ext.proto
 对客户端（Android设备，IOS设备）提供的rpc协议
 3.logic_server.ext.proto
 对业务服务器提供的rpc协议
 4.logic.int.proto
 对conn服务层提供的rpc协议
 5.conn.int.proto
 对logic服务层提供的rpc协议
 
项目目录简介
 
项目结构遵循 https://github.com/golang-standards/project-layout
 
api:          服务对外提供的grpc接口
cmd:          服务启动入口
config:       服务配置
internal:     每个服务私有代码
pkg:          服务共有代码
sql:          项目sql文件
test:         长连接测试脚本
 
服务简介
 
1.tcp_conn
 维持与客户端的TCP长连接，心跳，以及TCP拆包粘包，消息编解码
 2.ws_conn
 维持与客户端的WebSocket长连接，心跳，消息编解码
 3.logic
 设备信息，用户信息，群组信息管理，消息转发逻辑
 
TCP拆包粘包
 
遵循LV的协议格式，一个消息包分为两部分，消息字节长度以及消息内容。
 这里为了减少内存分配，拆出来的包的内存复用读缓存区内存。
 拆包流程：
 1.首先从系统缓存区读取字节流到buffer
 2.根据包头的length字段，检查报的value字段的长度是否大于等于length
 3.如果大于，返回一个完整包（此包内存复用），重复步骤2
 4.如果小于，将buffer的有效字节前移，重复步骤1
 
单用户多设备支持，离线消息同步
 
每个用户都会维护一个自增的序列号，当用户A给用户B发送消息是，首先会获取A的最大序列号，设置为这条消息的seq，持久化到用户A的消息列表，
 再通过长连接下发到用户A账号登录的所有设备，再获取用户B的最大序列号，设置为这条消息的seq，持久化到用户B的消息列表，再通过长连接下发
 到用户B账号登录的所有设备。
 假如用户的某个设备不在线，在设备长连接登录时，用本地收到消息的最大序列号，到服务器做消息同步，这样就可以保证离线消息不丢失。
 
度扩散和写扩散
 
首先解释一下，什么是读扩散，什么是写扩散
 
读扩散
 
简介：群组成员发送消息时，先建立一个会话，都将这个消息写入这个会话中，同步离线消息时，需要同步这个会话的未同步消息
 优点：每个消息只需要写入数据库一次就行，减少数据库访问次数，节省数据库空间
 缺点：一个用户有n个群组，客户端每次同步消息时，要上传n个序列号，服务器要对这n个群组分别做消息同步
 
写扩散
 
简介：在群组中，每个用户维持一个自己的消息列表，当群组中有人发送消息时，给群组的每个用户的消息列表插入一条消息即可
 优点：每个用户只需要维护一个序列号和消息列表
 缺点：一个群组有多少人，就要插入多少条消息，当群组成员很多时，DB的压力会增大
 
消息转发逻辑选型以及特点
 
普通群组：
 
采用写扩散，群组成员信息持久化到数据库保存。支持消息离线同步。
 
超大群组：
 
采用读扩散，群组成员信息保存到redis,不支持离线消息同步。
 
核心流程时序图
 
长连接登录
 
 
离线消息同步
 
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-TZANtEnF-1583670773426)(http://s1.wailian.download/2019/12/25/ef9c9452e65be3ced63573164fec7ed5.png)]
 
心跳
 
 
消息单发
 
c1.d1和c1.d2分别表示c1用户的两个设备d1和d2,c2.d3和c2.d4同理
 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KuLuB6jN-1583670773430)(http://www.wailian.work/images/2019/10/26/e000fda2f18e86f3.png)]
 
小群消息群发
 
c1,c2.c3表示一个群组中的三个用户
 
 
大群消息群发
 
 
错误处理,链路追踪,日志打印
 
系统中的错误一般可以归类为两种，一种是业务定义的错误，一种就是未知的错误，在业务正式上线的时候，业务定义的错误的属于正常业务逻辑，不需要打印出来，
 但是未知的错误，我们就需要打印出来，我们不仅要知道是什么错误，还要知道错误的调用堆栈，所以这里我对GRPC的错误进行了一些封装，使之包含调用堆栈。
 
func WrapError(err error) error {
	if err == nil {
		return nil
	}

	s := &spb.Status{
		Code:    int32(codes.Unknown),
		Message: err.Error(),
		Details: []*any.Any{
			{
				TypeUrl: TypeUrlStack,
				Value:   util.Str2bytes(stack()),
			},
		},
	}
	return status.FromProto(s).Err()
}
// Stack 获取堆栈信息
func stack() string {
	var pc = make([]uintptr, 20)
	n := runtime.Callers(3, pc)

	var build strings.Builder
	for i := 0; i < n; i++ {
		f := runtime.FuncForPC(pc[i] - 1)
		file, line := f.FileLine(pc[i] - 1)
		n := strings.Index(file, name)
		if n != -1 {
			s := fmt.Sprintf(" %s:%d \n", file[n:], line)
			build.WriteString(s)
		}
	}
	return build.String()
}
 
这样，不仅可以拿到错误的堆栈，错误的堆栈也可以跨RPC传输，但是，但是这样你只能拿到当前服务的堆栈，却不能拿到调用方的堆栈，就比如说，A服务调用
 B服务，当B服务发生错误时，在A服务通过日志打印错误的时候，我们只打印了B服务的调用堆栈，怎样可以把A服务的堆栈打印出来。我们在A服务调用的地方也获取
 一次堆栈。
 
func WrapRPCError(err error) error {
	if err == nil {
		return nil
	}
	e, _ := status.FromError(err)
	s := &spb.Status{
		Code:    int32(e.Code()),
		Message: e.Message(),
		Details: []*any.Any{
			{
				TypeUrl: TypeUrlStack,
				Value:   util.Str2bytes(GetErrorStack(e) + " --grpc-- \n" + stack()),
			},
		},
	}
	return status.FromProto(s).Err()
}

func interceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	err := invoker(ctx, method, req, reply, cc, opts...)
	return gerrors.WrapRPCError(err)
}

var LogicIntClient   pb.LogicIntClient

func InitLogicIntClient(addr string) {
	conn, err := grpc.DialContext(context.TODO(), addr, grpc.WithInsecure(), grpc.WithUnaryInterceptor(interceptor))
	if err != nil {
		logger.Sugar.Error(err)
		panic(err)
	}

	LogicIntClient = pb.NewLogicIntClient(conn)
}
 
像这样，就可以获取完整一次调用堆栈。
 错误打印也没有必要在函数返回错误的时候，每次都去打印。因为错误已经包含了堆栈信息
 
// 错误的方式
if err != nil {
	logger.Sugar.Error(err)
	return err
}

// 正确的方式
if err != nil {
	return err
}
 
然后，我们在上层统一打印就可以
 
func startServer {
    extListen, err := net.Listen("tcp", conf.LogicConf.ClientRPCExtListenAddr)
    if err != nil {
    	panic(err)
    }
	extServer := grpc.NewServer(grpc.UnaryInterceptor(LogicClientExtInterceptor))
	pb.RegisterLogicClientExtServer(extServer, &LogicClientExtServer{})
	err = extServer.Serve(extListen)
	if err != nil {
		panic(err)
	}
}

func LogicClientExtInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
	defer func() {
		logPanic("logic_client_ext_interceptor", ctx, req, info, &err)
	}()

	resp, err = handler(ctx, req)
	logger.Logger.Debug("logic_client_ext_interceptor", zap.Any("info", info), zap.Any("ctx", ctx), zap.Any("req", req),
		zap.Any("resp", resp), zap.Error(err))

	s, _ := status.FromError(err)
	if s.Code() != 0 && s.Code() < 1000 {
		md, _ := metadata.FromIncomingContext(ctx)
		logger.Logger.Error("logic_client_ext_interceptor", zap.String("method", info.FullMethod), zap.Any("md", md), zap.Any("req", req),
			zap.Any("resp", resp), zap.Error(err), zap.String("stack", gerrors.GetErrorStack(s)))
	}
	return
}
 
这样做的前提就是，在业务代码中透传context,golang不像其他语言，可以在线程本地保存变量，像Java的ThreadLocal,所以只能通过函数参数的形式进行传递，gim中，service层函数的第一个参数
 都是context，但是dao层和cache层就不需要了，不然，显得代码臃肿。
 最后可以在客户端的每次请求添加一个随机的request_id，这样客户端到服务的每次请求都可以串起来了。
 
func getCtx() context.Context {
	token, _ := util.GetToken(1, 2, 3, time.Now().Add(1*time.Hour).Unix(), util.PublicKey)
	return metadata.NewOutgoingContext(context.TODO(), metadata.Pairs(
		"app_id", "1",
		"user_id", "2",
		"device_id", "3",
		"token", token,
		"request_id", strconv.FormatInt(time.Now().UnixNano(), 10)))
}
 
github
 
https://github.com/alberliu/gim

公司原有的业务消息推送是靠前端 ajax 轮眉请求后端接口完成的。然后我新来的，让我改成 websocket 双向通信的来做消息推送。
 
简单的业务场景就是做一个聊天室，mq转发消息，在php bin/hyperf.php start 启动后，老是进程自动断掉：
 
 
一直找不到原因，后来看到官方文档 ：https://hyperf.wiki/2.0/#/zh-cn/quick-start/install
 
说下面的这些扩展不兼容。
 
存在兼容性问题的扩展
 
由于 Hyperf 基于 Swoole 协程实现，而 Swoole 4 带来的协程功能是 PHP 前所未有的，所以与不少扩展都仍存在兼容性的问题。
 以下扩展（包括但不限于）都会造成一定的兼容性问题，不能与之共用或共存：
 
xhprof
xdebug
blackfire
trace
uopz
我检查了一下我的扩展，没有这些，那会是啥情况啊，swoole 从4.6.7 到 4.6.0 全部安装了一遍，还是这个问题，连接刚一启动就断掉。
 
我再次检查一下安装的扩展。
 
 
注意到有个grpc.
 
gRPC 是一个高性能、开源和通用的 RPC 框架，面向移动和 HTTP/2 设计。目前提供 C、Java 和 Go 语言版本，分别是：grpc, grpc-java, grpc-go. 其中 C 版本支持 C, C++, Node.js, Python, Ruby, Objective-C, PHP 和 C# 支持.
 
gRPC 基于 HTTP/2 标准设计，带来诸如双向流、流控、头部压缩、单 TCP 连接上的多复用请求等特。这些特性使得其在移动设备上表现更好，更省电和节省空间占用。于是果断禁掉！
 
 
 
接着重新打开hyperf服务，正常了！
 
 
 
 
 
 
grpc底层用的是http2来连接的，通过node.js做服务的，所以可能是这个原因导致hyperf关闭！
 
 
 
 

背景：

locust默认内部只封装httplocust；使用的是requests中的session进行了封装；如果我想测试其它协议怎么办，比如websocket  , grpc；我们只要重写一个实例给client即可：

重写WebSocketClient类（主要用来替换掉self.client的http实例）
class WebSocketClient(object):

    def __init__(self, host):
        self.host = host
        self.ws = websocket.WebSocket()

    def connect(self, burl):
        start_time = time.time()
        try:
            self.conn = self.ws.connect(url=burl)
        except websocket.WebSocketTimeoutException as e:
            total_time = int((time.time() - start_time) * 1000)
            events.request_failure.fire(request_type="websockt", name='urlweb', response_time=total_time, exception=e)
        else:
            total_time = int((time.time() - start_time) * 1000)
            events.request_success.fire(request_type="websockt", name='urlweb', response_time=total_time, response_length=0)
        return self.conn

    def recv(self):
        return self.ws.recv()

    def send(self, msg):
        self.ws.send(msg)

注意：该类中定义了，websocket的常用操作，链接、接收、发送；最主要是events.request_failure.fire和events.request_success.fire这两个用来收集性能数据，如果不写报告收集不到性能数据


2、重写一个HttpLocust类，我们这里叫做WebsoketLoscust类
class WebsocketLocust(Locust):
    def __init__(self, *args, **kwargs):
        super(WebsocketLocust, self).__init__(*args, **kwargs)
        self.client = WebSocketClient(self.host)

注意：WebsocketLocust从Locust继承； 这里主要是将self.client重新实例成，我们第一部写好的websocketClient实例


3、编写TaskSet类

class SupperDianCan(TaskSet):

    @task
    def test_baidu(self):
        self.url = 'wss://xxxxxx.xxxx.com/cart/chat?sid=11303&table_no=103&user_id=ofZjWs40HxEzvV08l6m4PnqGbxqc_2_1_&version=2'

        self.data = {}

        self.client.connect(self.url)
        while True:
            recv = self.client.recv()
            print(recv)
            if eval(recv)['type'] == 'keepalive':
                self.client.send(recv)
            else:
                self.client.send(self.data)

注意：此类就是任务类，跟http的写法一样，只是这里用的self.client.xxxx已经变成了我们自已重写的websocket类，将原来的requests http替换了


4/编写站点类
class WebsiteUser(WebsocketLocust):

    task_set = SupperDianCan

    min_wait=5000

    max_wait=9000

注意：站点类从第二步中的locust继承
 
完整代码1：
 
from locust import Locust, events, task, TaskSet

import websocket

import time

import gzip

 

class WebSocketClient():

     def __init__(self, host):

         self.host = host

         #self.port = port

 

class WebSocketLocust(Locust):

     def __init__(self, *args, **kwargs):

         self.client = WebSocketClient("1xx.xx.xx.85")

 

class UserBehavior(TaskSet):



 

     @task(1)

     def buy(self):

         try:

            ws = websocket.WebSocket()

             # self.ws.connect("ws://xx:8807")

             ws.connect("ws://xxxx.com/r1/xx/ws")

 

             start_time = time.time()

 

             #self.ws.send('{"url":"/buy","data":{"id":"123","issue":"20170822","doubled_num":2}}')

            #result = self.ws.recv()

 

            send_info = '{"sub": "market.ethusdt.kline.1min","id": "id10"}'

             # send_info = '{"event":"subscribe", "channel":"btc_usdt.deep"}'

             while True:

                 # time.sleep(5)

                # ws.send(json.dumps(send_info))

                 ws.send(send_info)

                 while (1):

                    compressData = ws.recv()

                    result = gzip.decompress(compressData).decode('utf-8')

                     if result[:7] == '{"ping"':

                        ts = result[8:21]

                         pong = '{"pong":' + ts + '}'

                         ws.send(pong)

                        ws.send(send_info)

                     # else:

                    #     # print(result)

                    #     with open('./test_result.txt', 'a') as f:

                    #         #f.write(threading.currentThread().name + '\n')

                    #         f.write(result + '\n')

         except Exception as e:

             print("error is:",e)

 

class ApiUser(WebSocketLocust):

    task_set = UserBehavior

     min_wait = 100

     max_wait = 200
 
完整代码2：
 
# -*- encoding:utf-8 -*-

import gzip
import json
import random
import threading
import time
import zlib
from threading import Timer

import websocket
from gevent._semaphore import Semaphore
from locust import TaskSet, task, Locust, events

# TODO: 设置集合点...
all_locusts_spawned = Semaphore()
all_locusts_spawned.acquire()


def on_hatch_complete(**kwargs):
    all_locusts_spawned.release()


events.hatch_complete += on_hatch_complete

t2 = 0
repCount = 0
sendCount = 0
pingCount = 0
stSend = 0
openTime = 0
reqLen = 0
recordSt = 0
repList = []
printCount = 1
reqSentCount = 1

symbols = ["etcusdt"]

subbedCount = 0
retSubTopicCount = 0
testFlag = 0

def on_message(ws, message):
    global t2
    global repCount
    global sendCount
    global pingCount
    global stSend
    global printCount
    global reqList
    global recordSt
    global subbedCount
    global retSubTopicCount
    global reqSentCount

    req_list = {

        "req_str1": '{"req": "market.%s.kline.1min"}' % random.choice(symbols),
        "req_str2": '{"req": "market.%s.depth.step0"}' % random.choice(symbols),
        "req_str3": '{"req": "market.%s.trade.detail"}' % random.choice(symbols),
        "req_str4": '{"req": "market.%s.detail"}' % random.choice(symbols),
        # "req_str5": '{"req": "market.overview"}',

    }



    # 对返回来的压缩数据进行解压缩
    ws_result = zlib.decompressobj(31).decompress(message)

    result = json.loads(ws_result.decode('utf-8'))
    print(result)

    recordEd = time.time()  # 为了判断什么时候统计数据的结束时间

    recordCost = round((recordEd - recordSt) * 1000, 3)  # 统计的结束时间减去统计的开始时间

    # print(result)

    if 'subbed' in result:

        subbedCount = subbedCount + 1

        if subbedCount % 5 == 0:
            print("----------------subbed all topic----------------")

    if 'ch' in result:
        retSubTopicCount = retSubTopicCount + 1

    if 'rep' in result:
        repCount = repCount + 1

        repRetTime = int((time.time() - stSend) * 1000)

        repList.append(repRetTime)

        # print("the server rep time is ---->%dms" % repRetTime)

        # print("the server rep data is ---->%s" % result)



    # 判断ping的返回 ，对应给服务器发送pong
    if 'ping' in result:
        pingCount = pingCount + 1
        ping_id = result.get('ping')
        pong_str = '{"pong": %d}' % ping_id
        ws.send(pong_str)

        t1 = ping_id

        t3 = ping_id - t2

        t2 = t1

        if t3 > 5000:
            print("$$$$$$$time difference ping is %d$$$$$$$ " % t3)

        # print("ret ping value %d" % ping_id)
        # print("ret ping curTime %d" % int(time.time()*1000))
        # if 1000 < int((time.time()*1000) - ping_id):
        #     print("cur - pingTime is  ---> %dms" % int((time.time()*1000) - ping_id))

    if recordCost >= (random.randint(2000, 3000) * reqSentCount):
        reqSentCount += 1
        for key in req_list.keys():
            ws.send(req_list[key])

            sendCount = sendCount + 1

            # print("send  req info is --------->", req_list[key])

            stSend = time.time()

        # print("**********send count is %d   *************** " % sendCount)

    # 每1分钟统计一次
    if recordCost >= (60000 * printCount):
        printCount = printCount + 1

        curTime = time.strftime('%Y-%m-%d %H:%M:%S')

        repList.sort()

        retCount = len(repList)

        writeData = '| 当前时间%s ,req发送条数%s,返回总数据条数%s |  数据95耗时:%s  | 数据50耗时:%s  | sub返回量:%s ' % (
        curTime, sendCount, repCount, repList[int(retCount * 0.95)], repList[int(retCount * 0.5)], retSubTopicCount)

        fid = open("GipRecord.txt", "a+")

        fid.write(writeData + "\n")

        fid.close()

# 重新实现对应事件
def on_error(ws, error):
    print("occur error " + error)


def on_close(ws):
    global printCount
    global reqSentCount
    printCount = 1
    reqSentCount = 1
    print("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^con is closed^^^^^^^^^^^^^^^^^^^^^^^^^^^^^")


def on_open(ws):
    print("con success ...")
    global reqList
    global sendCount
    global reqLen
    global recordSt
    global stSend

    recordSt = time.time()  # 为了统计记录文件创建的开始时间

    stSend = time.time()

    sub_list = {

        "sub_str1": '{"sub": "market.%s.kline.1min"}' % random.choice(symbols),
        "sub_str2": '{"sub": "market.%s.depth.step0"}' % random.choice(symbols),
        "sub_str3": '{"sub": "market.%s.trade.detail"}' % random.choice(symbols),
        "sub_str4": '{"sub": "market.%s.detail"}' % random.choice(symbols),
        "sub_str5": '{"sub": "market.overview"}',

    }

    for key in sub_list.keys():
        ws.send(sub_list[key])


class WSClient(object):

    def __init__(self, host):
        self.ws = None
        self.host = host

    def useWebCreate(self):
        # websocket.enableTrace(True)
        self.ws = websocket.WebSocketApp(self.host,
                                         # header={'cloud-exchange':'510a02991'},
                                         on_message=on_message,
                                         on_error=on_error,
                                         on_close=on_close,
                                         on_open=on_open)

    def execute(self):
        self.ws.run_forever()


class AbstractLocust(Locust):
    def __init__(self, *args, **kwargs):
        super(AbstractLocust, self).__init__(*args, **kwargs)
        self.client = WSClient(self.host)


class ApiUser(AbstractLocust):
    host = 'ws://xxx/ws'
   
    min_wait = 10
    max_wait = 1000

    class task_set(TaskSet):
        def on_start(self):
            self.client.useWebCreate()
            # TODO: 设置集合点...
            all_locusts_spawned.wait()

        @task
        def execute_long_run(self):
            self.client.execute()
 
完整代码3
 
# -*- encoding:utf-8 -*-

import websocket
import threading
import time
import zlib
import json
from locust import TaskSet, task, User, between



t2 = 0
repCount = 0
sendCount = 0
pingCount = 0
stSend = 0
openTime = 0
reqLen = 0
recordSt = 0
repList = []
printCount = 1


def on_message(ws, message):

    global repCount
    global sendCount
    global pingCount
    global stSend
    # 对返回来的压缩数据进行解压缩
    ws_result = zlib.decompressobj(31).decompress(message)

    result = json.loads(ws_result.decode())

    # print('result->', result)   # 调试使用

    if 'rep' in result:

        repCount = repCount + 1

        retCost = int(round(time.time() * 1000)) - int(result["id"])   # 收到数据的系统时间 - 收到数据里的id

        # print("##########send count is  %d   #############" % sendCount)
        print("##########rep  count is  %d   #############" % repCount)

        if retCost >= 1000:

            curTime = time.strftime('%Y-%m-%d %H:%M:%S')

            writeData = '| 当前时间%s ,发送条数%s,返回总数据条数%s |  耗时:%s毫秒  |' % (curTime, sendCount, repCount, retCost)

            fid = open("reqRecord.txt", "a+")

            fid.write(writeData + "\n")

            fid.close()

    # 判断ping的返回 ，对应给服务器发送pong
    if 'ping' in result:

        pingCount = pingCount+1

        ping_id = result.get('ping')

        pong_str = '{"pong": %d}' % ping_id

        ws.send(pong_str)

# 重新实现对应事件
def on_error(ws, error):
    print("occur error " + error)


def on_close(ws):
    print("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^con is closed^^^^^^^^^^^^^^^^^^^^^^^^^^^^^")


class newRun(threading.Thread):

    def __init__(self, ws):

        threading.Thread.__init__(self)
        self.ws = ws

    def run(self):

        global sendCount
        global printCount

        st = time.time()

        while True:

            time.sleep(1)

            sendCount = sendCount + 1

            # everyReqList = ['{"req": "market.etc11pbtc.kline.1min","id":"%d","from":1616774400,"to":1616817600}' % int(round(time.time()*1000))]
            everyReqList = ['{"req": "market.etc11pbtc.kline.1min","id":"%d","from":1620717333,"to":1620718333}' % int(
                round(time.time() * 1000)),
                            '{"req": "market.etc12pbtc.kline.5min","id":"%d","from":1620717333,"to":1620718333}' % int(
                                round(time.time() * 1000)),
                            '{"req": "market.etc1pbtc.kline.5min","id":"%d","from":1620717333,"to":1620718333}' % int(
                                round(time.time() * 1000)),
                            '{"req": "market.etc2pbtc.kline.5min","id":"%d","from":1620717333,"to":1620718333}' % int(
                                round(time.time() * 1000)),
                            # '{"req": "market.etc11pbtc.trade.detail","id":"%d"}' % int(
                            #     round(time.time() * 1000))
                            ]
            for k in range(0,4):
                self.ws.send(everyReqList[k])

            ct = int((time.time() - st) * 1000)

            if ct >= printCount * 1000:
                print("=====send count is %d====" % sendCount)
                printCount = printCount + 1



def on_open(ws):

    print("con success ...")

    t1 = newRun(ws)

    t1.start()



class WSClient(object):

    def __init__(self, host):
        self.ws = None
        self.host = host


    def useWebCreate(self):
        self.ws = websocket.WebSocketApp(self.host,
                                    on_message=on_message,
                                    on_error=on_error,
                                    on_close=on_close,
                                    on_open=on_open)

    def execute(self):

        self.ws.run_forever()


class AbstractLocust(User):
    def __init__(self, *args, **kwargs):
        super(AbstractLocust, self).__init__(*args, **kwargs)
        self.host = 'ws://xxxx.com:80/ws'
        self.client = WSClient(self.host)

    wait_time = between(0, 1)

    @task
    class task_set(TaskSet):
        def on_start(self):
            self.client.useWebCreate()

        @task
        def execute_long_run(self):
            self.client.execute()
 
完整代码4
 
# -*- encoding:utf-8 -*-

import base64
import configparser
import hashlib
import hmac
import json
import random
import time
from datetime import datetime
from urllib import parse
import linecache
import random
from locust import HttpUser

import websocket
import xlrd
from gevent._semaphore import Semaphore
# from locust import TaskSet, task, Locust,events
from locust import TaskSet, task, User,between

# # TODO: 设置集合点...
# all_locusts_spawned = Semaphore()
# all_locusts_spawned.acquire()
#
#
# def on_hatch_complete(**kwargs):
#     all_locusts_spawned.release()

#
# events.hatch_complete += on_hatch_complete

conf = configparser.ConfigParser()
# # conf.read("domain.ini")
# # domain = conf.get('DevSet','ip')
domain = 'user-data-push.loadtest-5.hk3.huobiapps.com'
# file_name = "tokenads.txt"

t2 = 0
repCount = 0
sendCount = 0
pingCount = 0
subRepCount = 0
row = 0
recordSt = 0
printCount = 1
subOneMinCount = 0
subSymbolCount = 0
sdTime = 0
endTime = 0
req2002 = 0
startResultTime = 0
endResultTime = 0
resultTimeList = []




# print("+++++++++" + domain)
def on_message(ws, message):
    global t2
    global startResultTime
    global endResultTime
    global repCount
    global sendCount
    global pingCount
    global subRepCount
    global recordSt
    global printCount
    global subOneMinCount
    global subSymbolCount
    global endTime
    global if_write
    global req2002
    global resultTimeList

    # 对返回来的压缩数据进行解压缩

    result = json.loads(message)

    print("---- %s" % result)


    sub_list = {
        "sub_str1": '{"action": "sub","ch": "accounts.update#1"}',
        "sub_str2": '{"action": "sub","ch": "trade.clearing#*#0"}',#成交
        "sub_str4": '{"action": "sub","ch": "trade.clearing#*#1"}',
        "sub_str3": '{"action": "sub","ch": "orders#*"}',  #
        # "sub_str2": '{"action": "sub","ch": "trade.clearing#*#0"}',  # 成交
        # "sub_str4": '{"action": "sub","ch": "trade.clearing#*#1"}',
        # "sub_str2": '{"action": "sub","ch": "trade.clearing#*#0"}',  # 成交
        # "sub_str4": '{"action": "sub","ch": "trade.clearing#*#1"}',
    }

    # 判断ping的返回 ，对应给服务器发送pong
    if 'action' in result:
        if 'ch' in result:
            if result['ch'] == 'auth' and result["code"] == 200:
                print("auth success")

                for key in sub_list.keys():
                    startResultTime = time.time()
                    # print("发送前时间是 %s" % startResultTime)
                    a = random.sample(sub_list.keys(), 1)
                    print("************",a)
                    # print(a)
                    ws.send(sub_list[a[0]])
                    break
                    # ws.send(sub_list[key])
                    # print(sub_list)

        else:
            pingCount = pingCount + 1

            ping_id = result['data']['ts']

            # print(ping_id)

            pong_str = '{"action":"pong","params":{"ts": %d}}' % ping_id

            ws.send(pong_str)
            # print("send pong info is -->" + pong_str)

            t1 = ping_id

            t3 = ping_id - t2

            t2 = t1

            if t3 > 5000:
                print("$$$$$$$time difference ping is %d$$$$$$$ " % t3)

        # if subRepCount % 12 == 0:
        #     print("Receive  sub count is %d" % subRepCount)

    # 每1分钟统计一次
    recordEd = time.time()
    recordCost = round((recordEd - recordSt) * 1000, 3)  # 统计的结束时间减去统计的开始时间

    if recordCost >= (60000 * 5 * printCount):
        printCount = printCount + 1
        resultTimeList.sort()
        retCount = len(resultTimeList)
        avg_time = round(sum(resultTimeList) / retCount)

        curTime = time.strftime('%Y-%m-%d %H:%M:%S')

        writeData = '| 当前时间%s ,当前存在%s条数据,|平均耗时 : %s | 数据70耗时:%s  |   数据95耗时:%s  | 数据99耗时:%s  |' % (
            curTime, retCount, avg_time, resultTimeList[int(retCount * 0.7)], resultTimeList[int(retCount * 0.95)],
            resultTimeList[int(retCount * 0.99)])


        fid = open("account-record.txt", "a+")

        fid.write(writeData + "\n")

        fid.close()

        # 每分钟重置一次
        subOneMinCount = 0


# 重新实现对应事件
def on_error(ws, error):
    print("occur error %s" % error)


def on_close(ws):
    global printCount
    printCount = 1
    print("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^con is closed^^^^^^^^^^^^^^^^^^^^^^^^^^^^^")


def on_open(ws):
    global recordSt
    global sdTime

    recordSt = time.time()

    print("con success ...")
    # # 加签
    file_name = "loadtest-5-UserDataPush"
    # count = len(open(file_name, 'rU').readlines())  # 获取行数
    count = len(open(file_name, 'rU').readlines())  # 获取行数
    print("当前有%s行 " % count)
    random_col = random.randrange(1, count, 1)
    print("当前随机取到%s行" % random_col)
    line = linecache.getline(file_name, random_col)
    print(line)
    data = line.strip().split(",")
    access_key = data[0]
    secret_key = data[1]


    ##test2
    # access_key = "364ed29f-nbtycf4rw2-2c67cc5d-e2814"
    # secret_key = "5b02fcf3-500608e9-b561af74-cfd8d"
    print("--------" + access_key)
    print("++++++" + secret_key)
    authData = [
        secret_key.encode('utf-8'),
        {
            "authType": "api",
            "accessKey": access_key,
            "signatureMethod": "HmacSHA256",
            "signatureVersion": "2.1",
            "timestamp": authfunc()._utc()
        },

    ]

    send_msg = authfunc()._auth(authData)

    sendData = {"action": "req", "ch": "auth", "params": send_msg}

    print("+++++++++")

    print(json.dumps(sendData))

    ws.send(json.dumps(sendData))


class WSClient(object):

    def __init__(self, host):
        self.ws = None
        self.host = host

    def useWebCreate(self):
        # websocket.enableTrace(True)
        self.ws = websocket.WebSocketApp(self.host,
                                         on_message=on_message,
                                         on_error=on_error,
                                         on_close=on_close,
                                         on_open=on_open,
                                         header=["X-HB-Exchange-Code:pro"])
    def execute(self):
        self.ws.run_forever()




class AbstractLocust(User):
    def __init__(self, *args, **kwargs):
        super(AbstractLocust, self).__init__(*args, **kwargs)
        domain='xx.huobiapps.com'
        self.host = 'ws://' + domain + '/ws/v2'
        self.client = WSClient(self.host)
        wait_time = between(0, 1)

        # print("%s" %self.client)




# class ApiUser(AbstractLocust):
# class ApiUser(AbstractLocust):
#
#     host = 'ws://' + domain + '/ws/v2'
#     print("++++++host is %s" % host)
#     min_wait = 10
#     max_wait = 1000
#     wait_time = between(0, 1)

    @task
    class task_set(TaskSet):
        def on_start(self):
            self.client.useWebCreate()
            # TODO: 设置集合点...
            # all_locusts_spawned.wait()

        @task
        def execute_long_run(self):
            self.client.execute()


# 加签
class authfunc(object):

    def _sign(self, param=None, _accessKeySecret=None):
        # create signature:
        if param is None:
            params = {}
        params = {}
        # params['signatureMethod'] = param.get('signatureMethod') if type(param.get('signatureMethod')) == type(
        #     'a') else '' if param.get('signatureMethod') else ''
        # params['signatureVersion'] = param.get('signatureVersion') if type(param.get('signatureVersion')) == type(
        #     'a') else '' if param.get('signatureVersion') else ''
        # params['accessKey'] = param.get('accessKey') if type(param.get('accessKey')) == type(
        #     'a') else '' if param.get('accessKey') else ''
        # params['timestamp'] = param.get('timestamp') if type(param.get('timestamp')) == type('a') else '' if param.get(
        #     'timestamp') else ''

        params['signatureMethod'] = param.get('signatureMethod') if type(param.get('signatureMethod')) == type(
            'a') else '' if param.get('signatureMethod') else ''
        params['signatureVersion'] = param.get('signatureVersion') if type(param.get('signatureVersion')) == type(
            'a') else '' if param.get('signatureVersion') else ''
        params['accessKey'] = param.get('accessKey') if type(param.get('accessKey')) == type(
            'a') else '' if param.get('accessKey') else ''
        params['timestamp'] = param.get('timestamp') if type(param.get('timestamp')) == type('a') else '' if param.get(
            'timestamp') else ''

        # print(params)
        # sort by key:
        keys = sorted(params.keys())
        _host = domain
        path = '/ws/v2'
        # build query string like: a=1&b=%20&c=:
        qs = '&'.join(['%s=%s' % (key, self._encode(params[key])) for key in keys])
        # build payload:
        payload = '%s\n%s\n%s\n%s' % ('GET', _host, path, qs)
        # print(payload)
        # print('payload:\n%s' % payload)
        dig = hmac.new(_accessKeySecret, msg=payload.encode('utf-8'), digestmod=hashlib.sha256).digest()

        return base64.b64encode(dig).decode()

    def _encode(self, s):
        # return urllib.pathname2url(s)
        return parse.quote(s, safe='')

    def _utc(self):
        return datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S')

    def _auth(self, auth):
        authenticaton_data = auth[1]
        _accessKeySecret = auth[0]

        authenticaton_data['signature'] = self._sign(authenticaton_data, _accessKeySecret)

        # print(authenticaton_data)

        return authenticaton_data
 
 
 
 
 
然后通过locust命令执行
 
locust -f xx.py  --no-web -c 2 -r 1 -t 1m