C++调用Go语言的CloudEvents SDK实现FaaS框架Knative Eventing中的证券行情服务

大多数证券行情服务的SDK都是C++版本的最成熟，本文示例了使用证券行情的Linux C++ SDK调用封装了CloudEvents的Go SDK的C++动态库，实现向Knative Eventing平台提供行情事件。代码为Demo性质，最终交付为容器镜像形式。
以此方式，使用任何开发语言都可以直接消费实时证券行情数据，对于量化研究场景，可直接使用Python获取行情，并实现计算任务。但行情经过FaaS引擎传递一定会消耗时间，所以此方式不适合对行情时效有极限要求的场景。代码分为两部分：使用Go语言为开发的C++动态库、使用C++开发的行情服务。

CloudEvents的C++ SDK

• 建议在WSL中进行开发，所以首先依官方文档安装WSL（安装默认的Ubuntu即可）
• 在开始菜单中，运行Ubuntu应用程序，启动shell窗口，配置Go开发环境

export GOPROXY=https://goproxy.cn,direct

sudo apt update
sudo apt install golang-go
mkdir cloudevents-sdk-cpp
cd cloudevents-sdk-cpp
vi event_sender.go

• 将以下代码输入event_sender.go

package main

import (
	"C"
	"context"
	"log"

	cloudevents "github.com/cloudevents/sdk-go/v2"
	uuid "github.com/satori/go.uuid"
)
import "os"

func main() {
	Send("{a:1,b:2}")
}

//export Send
func Send(data string) bool {
	bytedata := make([]byte, len(data))
	copy(bytedata, data)

	k_sink := os.Getenv("K_SINK")
	if k_sink == "" {
		log.Printf("failed to read sink address")
		return false
	}

	p, err := cloudevents.NewHTTP(cloudevents.WithTarget(k_sink))
	if err != nil {
		log.Printf("failed to create http protocol: %s", err.Error())
		return false
	}

	c, err := cloudevents.NewClient(p, cloudevents.WithUUIDs(), cloudevents.WithTimeNow())
	if err != nil {
		log.Printf("failed to create client: %s", err.Error())
		return false
	}

	event := cloudevents.NewEvent("1.0")
	event.SetType("com.quatation.data.received")
	event.SetSource("https://quatation-service.io")
	event.SetData(cloudevents.ApplicationJSON, bytedata)
	event.SetID(uuid.NewV4().String())

	log.Printf("sending cloudevent to %s", k_sink)
	if res := c.Send(context.Background(), event); !cloudevents.IsACK(res) {
		log.Printf("failed to send cloudevent: %v", res)
		return false
	}

	return true
}

• 编译成C++可调用的libevent_sender.so和event_sender.h

go mod init cloudevents-sdk-cpp
go get github.com/cloudevents/sdk-go/v2
go get github.com/satori/go.uuid
go build -o libevent_sender.so -buildmode=c-shared .
mv libevent_sender.h event_sender.h

CloudEvents格式的行情服务

• 下载安装Visual Studio 2022，安装时选择linux或嵌入设备的c++程序开发
• 启动Visual Studio 2022，选择Visual C++开发；创建新项目，选择CMake项目，名字为quotation-service-in-knative
• 创建include文件夹，放入rapidjson的头文件文件夹rapidjson，再放入上一步生成的event_sender.h和行情服务提供的头文件
• 创建lib文件夹，放入上一步生成的libevent_sender.so和行情服务提供的Linux C++ SDK的动态库文件
• 在quotation-service-in-knative.cpp中，输入如下代码

#include "quotation-service-in-knative.h"
#include "rapidjson/writer.h"
#include "rapidjson/stringbuffer.h"
#include "event_sender.h"
// 写入行情服务提供的头文件
#include ""

using namespace rapidjson;
using namespace std;

GoString BuildGoString(const char * p, size_t n) {
	return { p, static_cast<ptrdiff_t>(n) };
}

string Serialize(const /*行情数据结构名*/ * pQuotation)
{
	StringBuffer s;
	Writer<StringBuffer> writer(s);

	writer.StartObject();

	writer.Key(""); //写入行情数据结构里的字段名
	writer.Int(pQuotation->/*写入行情数据结构里的字段名*/);

	writer.Key(""); //写入行情数据结构里的字段名
	writer.String(pQuotation->/*写入行情数据结构里的字段名*/);

	writer.Key(""); //写入行情数据结构里的字段名
	writer.Double(pQuotation->/*写入行情数据结构里的字段名*/);

	writer.Key(""); //写入行情数据结构里的字段名
	writer.StartArray();
	for (unsigned i = 0; i < 10; i++)
		writer.Double(pQuotation->/*写入行情数据结构里的字段名*/[i]);
	writer.EndArray();
	writer.EndObject();

	return s.GetString();
};


class CSQSPIInFaaS : public /*行情SDK数据处理接口类名*/
{
public:
	void OnData(const /*行情数据结构名*/ * pQuotation)
	{
		string sData = Serialize(pQuotation);
		cout << sData << endl;
		Send(BuildGoString(sData.c_str(), sData.size()));
	}
};

int main()
{
	/*行情数据服务接口类*/ * iAPI = /*创建行情数据服务接口函数*/();

	/*行情服务链接信息类*/ * iInit = new /*行情服务链接信息类*/();
	// 需联系行情服务获取行情中心IP、端口、用户名和密码
	strcpy(iInit ->host, "");
	iInit ->port = 0;
	strcpy(iInit ->userId, "");
	strcpy(iInit ->password, "");

	CSQSPIInFaaS * iCSQSPIInFaaS = new CSQSPIInFaaS();


	/*行情订阅数据结构名*/ * iSubField = new /*行情订阅数据结构名*/();
	// 需联系行情服务获取具体参数设置
	strcpy(iSubField->strclass, "");
	strcpy(iSubField->category, "");
	strcpy(iSubField->exchId, "");
	strcpy(iSubField->stkCode, "");

	iAPI->Init(iInit);
	iAPI->Set(iCSQSPIInFaaS);
	iAPI->Start();
	iAPI->Sub(iSubField);

	char x;
	cin >> x;

	iAPI->Stop();
	iAPI->UnInit();
	delete iSubField;
	delete iCSQSPIInFaaS;
	delete iInit;

	return 0;
}

• 目标系统选择WSL:Ubuntu，在CMakeLists.txt中输入，然后按窗口上方提示，点击生成按钮，生成Makefile

cmake_minimum_required (VERSION 3.8)

project ("quotation-service-in-knative")
include_directories(include)
link_directories(lib)

add_executable (quotation-service-in-knative " .cpp" "quotation-service-in-knative.h")
target_link_libraries(quotation-service-in-knative /*行情服务动态库名称*/ event_sender)

if (CMAKE_VERSION VERSION_GREATER 3.12)
  set_property(TARGET quotation-service-in-knative PROPERTY CXX_STANDARD 20)
endif()

• 制作支持C++和Go语言运行环境的Ubuntu容器镜像，也可跳过此步，直接使用taikeguluer/ubuntu-4-cpp-and-go
  • 运行Ubuntu容器镜像

    docker run --name ubuntu_demo -itd ubuntu:20.04
    

  • 在Docker Descktop中选择ubuntu_demo容器的CLI

    apt update
    apt install golang-go
    mkdir /home/app
    

  • 在WSL命令窗口中，运行docker ps获取ubuntu_demo容器的ID，如fd4fe6d147c9，创建Docker image

    docker commit fd4fe6d147c9 taikeguluer/ubuntu-4-cpp-and-go
    

• 新建Dockerfile文件，并输入如下代码，其中FROM后可以修改为上一步制作的镜像：

FROM taikeguluer/ubuntu-4-cpp-and-go
ENV K_SINK 能接受CloudEvents格式事件的服务地址
WORKDIR /home/app
ADD ./lib/libCon.so /usr/lib
ADD ./lib/libCon2.so /usr/lib
ADD ./lib/libevent_sender.so /usr/lib
ADD ./lib/libQuotationAPI.so /usr/lib
ADD ./out/build/linux-debug/quotation-service-in-knative /home/app
CMD ./quotation-service-in-knative

• 点击F7或从菜单选择全部生成
• 在WSL窗口中，生成容器镜像

cd ~/.vs/quotation-service-in-knative
docker build -t your-docker-hub-id/quotation-service-in-knative .

• 如仅需在无Knative环境的前提下测试，启动容器即可

docker run --name quotation-service-in-knative -itd your-docker-hub-id/quotation-service-in-knative

• 如需在Knative中测试，请参考IMBroker的第3.1.5节