timerring

Apache Thrift RPC and Matchmaking Service

August 8, 2023 · 72 min read
Tutorial
Thrift | VPS
If you have any questions, feel free to comment below. Click the block can copy the code.
And if you think it's helpful to you, just click on the ads which can support this site. Thanks!

thrift 是一种接口描述语言和二进制通讯协议，它被用来定义和创建跨语言的服务。

Thrift 是 Facebook 于 2007 年开发的跨语言的 RPC（Remote Procedure Call，远程过程调用）服务框架，提供多语言的编译功能，并提供多种服务器工作模式；用户通过 Thrift 的 IDL（接口定义语言）来描述接口函数及数据类型，然后通过 Thrift 的编译环境生成各种语言类型的接口文件，用户可以根据自己的需要采用不同的语言开发客户端代码和服务器端代码,让不同语言构建的服务可以做到远程调用无缝对接。（它们的目的都是提供本地调用远程服务的能力）

RPC（Remote Procedure Call，远程过程调用）是一个计算机通信协议，此协议允许进程间通信。简单来说，当机器 A 上的进程调用机器 B 上的进程时，A 上的调用进程被挂起，而 B 上的被调用进程开始执行。调用方可以通过参数将信息传送给被调用方，然后可以通过被调用方传回的结果得到返回
大部分的RPC框架都遵循如下三个开发步骤：

  1. 定义一个接口说明文件：描述了对象(结构体)、对象成员、接口方法等一系列信息；
  2. 通过RPC框架所提供的编译器，将接口说明文件编译成具体的语言文件；
  3. 在客户端和服务器端分别引入RPC编译器所生成的文件，即可像调用本地方法一样调用服务端代码；
    RPC是一种C/S架构的服务模型，server端提供接口供client调用，client端向server端发送数据，server端接收client端的数据进行相关计算并将结果返回给client端。

thrift官网[1] -> Tutorial[2] -> tutorial.thrift[3]
Thrift使用C++进行编写，在安装使用的时候需要安装依赖，windows安装方式见官网即可。
安装方式: 官网安装方式[4]

  • windows 下载 thrift.exe – http://archive.apache.org/dist/thrift/0.9.1/thrift-0.9.1.exe
  • centos 执行 yum install thrift
  • 通过源码编译安装 https://github.com/apache/thrift

安装 thrift python 包:

pip install thrift

查看thrift版本： (推荐0.16.0版本)

thrift -version

下面以Ubuntu20.04为例展示安装过程：
（1）下载依赖库

sudo apt-get install automake bison flex g++ git libboost-all-dev libevent-dev libssl-dev libtool make pkg-config

（2）下载安装文件并解压

wget https://dlcdn.apache.org/thrift/0.16.0/thrift-0.16.0.tar.gz
tar -xvzf thrift-0.16.0.tar.gz

（3）安装

cd thrift-0.16.0
sudo ./configure
sudo make
sudo make install

thrift的跨语言特型 #

thrift通过一个中间语言IDL(接口定义语言)来定义RPC的数据类型和接口,这些内容写在以.thrift结尾的文件中,然后通过特殊的编译器来生成不同语言的代码,以满足不同需要的开发者,比如java开发者,就可以生成java代码,c++开发者可以生成c++代码,生成的代码中不但包含目标语言的接口定义,方法,数据类型,还包含有RPC协议层和传输层的实现代码。

Thrift IDL #

Thrift 采用 IDL（Interface Definition Language） 来定义通用的服务接口，然后通过 Thrift 提供的编译器，可以将服务接口编译成不同语言编写的代码，通过这个方式来实现跨语言的功能。

  • 通过命令调用Thrift提供的编译器将服务接口编译成不同语言编写的代码。
  • 这些代码又分为服务端和客户端，将所在不同进程(或服务器)的功能连接起来。

使用IDL对接口进行描述的thrift文件命名一般都是以“.thrift”作为后缀：XXX.thrift

thrift 服务分为服务提供方（server 端）和服务请求方（client 端）

通过 IDL 文件做到 server 与 client 的解耦。

service DemoService {
    string say();
}

上面就是一个最简单的 IDL 文件，他表示服务名为：DemoService 的服务中提供了一个名为 say 的方法，这个方法无参数传入，返回 string 类型。

生成代码文件：

thrift -r --gen <language> <Thrift filename>

例如： thrift -r --gen py demoservice.thrift
demoservice 就是存储我们上述所说的 DemoService 的文件，如果要生成其他语言的代码，则将 py 换成对应的目标语言，如 java、cpp、php 等。

thrift 的协议栈结构 #

thrift 是一种 c/s 的架构体系。在最上层是用户自行实现的业务逻辑代码.第二层是由 thrift 编译器自动生成的代码，主要用于结构化数据的解析，发送和接收。TServer 主要任务是高效的接受客户端请求，并将请求转发给 Processor 处理。Processor 负责对客户端的请求做出响应，包括 RPC 请求转发，调用参数解析和用户逻辑调用，返回值写回等处理。从 TProtocol 以下部分是 thirft 的传输协议和底层 I/O 通信。TProtocol 是用于数据类型解析的，将结构化数据转化为字节流给 TTransport 进行传输。TTransport 是与底层数据传输密切相关的传输层，负责以字节流方式接收和发送消息体，不关注是什么数据类型。底层 IO 负责实际的数据传输，包括 socket、文件和压缩数据流等。

thrift应用 #

Thrift类型系统包括预定义的基本类型（如bool , byte, double, string）、特殊类型(如binary: 未经过编码的字节流)、用户自定义结构体（看上去像C 语言的结构体）、容器类型（如list，set，map），以及异常和服务定义~ 注意，thrift不支持无符号整型，因为很多目标语言不存在无符号整型（如java）。

thrift基础数据类型 #

//The first thing to know about are types. The available types in Thrift are:
bool        //Boolean, one byte               bool
i8 (byte)   //Signed 8-bit integer            byte
i16         //Signed 16-bit integer           short
i32         //Signed 32-bit integer           int
i64         //Signed 64-bit integer           long long
double      //64-bit floating point value     double
string      //String                          string
binary      //Blob (byte array)               ByteBuffer


//容器（Container）
map<t1,t2>  //Map from one type to another
list<t1>    //Ordered list of one type
set<t1>     //Set of unique elements of one type

List<t1> ：一系列t1类型的元素组成的有序表，元素可以重复
Set<t1> ：一系列t1类型的元素组成的无序表，元素唯一
Map<t1,t2> ：key/value对（key的类型是t1且key唯一，value类型是t2）

例子：

struct Test{
    1: list<i32> intList;
    2: map<i32, User> users
}

映射类型赋值时，键值用冒号:隔开

map<i32, string> users = {1: "xxx", 2: "yyy"}

可像c++用typedef给结构体取别名

typedef map<i32, string> xxx

声明要转化的语言 #

Thrift中的命名空间同C++中的namespace和java中的package类似，它们均提供了一种组织（隔离）代码的方式。因为每种语言均有自己的命名空间定义方式（如python中有module），thrift允许开发者针对特定语言定义namespace。

声明命名空间的必要性：防止不同空间里的变量名重复。

/**
 * Thrift files can namespace, package, or prefix their output in various
 * target languages.
 */

namespace cl tutorial
namespace cpp tutorial
namespace d tutorial
namespace dart tutorial
namespace java tutorial
namespace php tutorial
namespace perl tutorial
namespace haxe tutorial
namespace netstd tutorial

//匹配系统我们用C++实现。
//语法:namespace 使用的语言 空间名称
namespace cpp match_dao

namespace java com.xtxxtx.test 转换成 package com.xtxxtx.test

结构体定义 #

Thrift结构体在概念上同C语言结构体类型—-一种将相关属性聚集（封装）在一起的方式。在面向对象语言中，thrift结构体被转换成类，在Java语言中，这等价于JavaBean的概念

注意:在struct定义结构体时需要对每个结构体成员用序号标识：“序号: ”。

/**
 * Structs are the basic complex data structures. They are comprised of fields
 * which each have an integer identifier, a type, a symbolic name, and an
 * optional default value.
 *
 * Fields can be declared "optional", which ensures they will not be included
 * in the serialized output if they aren't set.  Note that this requires some
 * manual management in some languages.
 */
struct Work {
  1: i32 num1 = 0,
  2: i32 num2,
  3: Operation op,
  4: optional string comment,
}

异常（Exception） #

异常在语法和功能上类似于结构体，只不过异常使用关键字exception而不是struct关键字声明。但它在语义上不同于结构体—当定义一个RPC服务时，开发者可能需要声明一个远程方法抛出一个异常。如:

exception MyException {
    1: string code;
    2: string message;
}

服务Service定义 #

一个服务包含一系列命名函数，每个函数包含一系列的参数以及一个返回类型。
在语法上，服务等价于定义一个接口或者纯虚抽象类~

注意：参数需要加上编号

格式如下，

service <name> {
  <returntype> <name> (<arguments>)
  [throws (<exceptions>)]
...
}

如

service  UserService {  
  // service中可以定义若干个服务，相当于Java Interface中定义的方法
  string sayHello(1:string name);
}

Typedef #

与 C++类似

typedef i32 MyInteger

Include #

Thrift允许thrift文件包含，相当于CPP中的include，Java中的import，如：

include "user.thrift"

namespace java com.xxx.tutorial.thrift.service  

service  UserService {  

  string sayHello(1:string name),

  bool saveUser(1:user.User user)
}

说明：
a． thrift文件名要用双引号包含，末尾没有逗号或者分号
b． 注意user前缀

例子2：

IDL文件中对所有接口函数的描述都放在service中，service的名字可以自己指定，该名字也将被用作生成的特定语言接口文件的名字，接口函数需要对参数使用序号标号，除最后一个接口函数外，要以“，”结束对函数的描述。

例如，下面一个IDL描述的Thrift文件（该Thrift文件的文件名为：test_service.thrift）的全部内容：

namespace java com.test.service

include "thrift_datatype.thrift"

service TestThriftService
{

    /*
      * value 中存放两个字符串拼接之后的字符串
    */
    thrift_datatype.ResultStr getStr(1:string srcStr1, 2:string srcStr2),

    thrift_datatype.ResultInt getInt(1:i32 val)

}

这里的TestThriftService就被用作生成的特定语言的文件名，例如我想用该Thrift文件生成一个java版本的接口文件，那么生成的java文件名就是：TestThriftService.java。

枚举Enum #

可以像C/C++那样定义枚举类型，如：

enum Gender {
    MALE,
    FEMALE,
    UNKONWN
}

thrift 项目实战 #

thrift 是一种接口描述语言，用简便的 thrift 语言写完以后，可以用命令快速生成任何语言的文件

用 thrift 实现一个基本的匹配系统 接口

服务分为三部分：分别是 game，match_system，save_server

  • game为match_client端，通过match.thrift接口向match_system完成添加用户和删除用户的操作
  • match_system由两部分组成，分别为match_server端和save_client端。match_server端负责接收match_client端的操作，将用户添加进匹配池，并且在匹配结束之后通过save.thrift接口将用户数据上传至另一个服务器
  • save_server用于接收match_system上传的匹配成功的用户信息

这个游戏的功能可能运行在一个或多个服务器(或进程)上，而thrift就是将不同服务器不同语言的功能连接起来。
图中的三个节点 (功能)是完全独立的，既可以在同一个服务器上，也可以在不同服务器上。每一个节点就是一个进程，每个进程可以使用不同的语言来实现。

基本框架 #

  1. 游戏应用端（Python3）
    • 客户端(match_client)：与 匹配系统服务器 的服务端交互
  2. 匹配系统服务器（C++, match_system）
    • 服务端(match_server)：与 游戏应用端 的客户端交互
    • 客户端(save_client)：与 数据存储服务器 的服务端交互
  3. 数据存储服务器（已经实现,save_server）
    • 服务端：与 匹配系统服务器 的客户端交互(save.thrift)

游戏匹配服务项目流程 #

  • 构建match.thrift接口
  • 通过match.thrift接口构建服务端和客户端
  • 先将服务端和客户端跑通，能完成基本的连接通信
  • 完成match.thrift的客户端需求
  • 构建save.thrift接口
  • 通过save.thrift接口构建服务端和客户端
  • 将客户端业务添加到match_system当中，将save.thrift服务端完成（本项目save.thrift服务端已经提前做好了）
  • 根据业务需求完善match_system

实现流程 #

初始化 git 仓库 #

mkdir thrift_lesson
cd thrift_lesson
git init

mkdir thrift        # 存放 thrift 源文件
mkdir game          # 实现 游戏应用端 的 客户端 功能
mkdir match_system  # 实现 匹配系统服务器 的 服务端 和 客户端 功能

touch readme.md
# 远程创建好git上的repo
git remote add xxx
git add .
git commit -m "init repo"
git push -u origin master

实现 游戏应用端 与 匹配系统服务器 的交互 #

创建 match.thrift 接口文件
用于实现 游戏应用端 与 匹配系统服务器 交互的 service

namespace cpp match_service //声明转换为C++语言

struct User {   //定义结构体 User
    1: i32 id, 
    2: string name,
    3: i32 score
}

service Match { //定义service服务
    /** 
     * user: 添加的用户信息
     * info: 附加信息
     * 在匹配池中添加一个名用户
     */  

    i32 add_user(1: User user, 2: string info),

    /** 
     * user: 删除的用户信息
     * info: 附加信息
     * 从匹配池中删除一名用户
     */  

    i32 remove_user(1: User user, 2: string info),
}

实现匹配系统服务器的服务端(match_server) #

运行 match.thrift 接口文件在匹配系统服务器端生成 C++ 源文件

进入 match_system 中新建文件夹 src （之后所有的 匹配服务器 源文件放在 src 下）

并在 src 文件夹中运行 thrift 脚本生成C++ 版本的文件

# 原始的thrift生成文件语法：thrift -r --gen <language> <Thrift filename>
mkdir match_system/src
cd match_system/src
thrift -r --gen cpp ../../thrift/match.thrift

# 将该文件夹重命名为 match_server（区别于之后要此处生成的client server）
# match_server 与 游戏应用端交互 ; client_server 与 数据存储服务器交互
mv gen-cpp match_server

# 把 Match_server.skeleton.cpp 移动到当前 src 目录下并重命名为 main.cpp
# 方便之后调试 main.cpp 文件，其他的源文件仍被存放在 src 文件夹下
mv match_server/Match_server.skeleton.cpp main.cpp

# 由于移动了 main.cpp 故需要修改一下 main.cpp 中头文件里关于 Math.cpp 的引用路径: match_server/Match.h
# 之后写 client_server 同理

具体操作如下所示：

写 thrift 服务的时候，先编译成功，然后再逐步向文件中添加模块。

编译并运行 cpp 文件 CPP编译流程： 1. 编译(g++ -c .cpp) 2. 链接(g++ -o *.o -o 可执行文件名 -动态链接库)

编译只需要编译.cpp文件，头文件(.h)不需要编译 将所有想要编译的文件，放在g++ -c命令后面即可
编译完成后会生成.o文件，有多少个.cpp文件就有多少个.o文件
最后将所有.o文件链接起来生成可执行文件即可。

# 1. 编译 所有的 .cpp 文件生成 .o 文件
# g++ -c [文件1.cpp] [文件2.cpp] ...
g++ -c main.cpp match_server/*.cpp

# 2. 链接 所有的 .o 文件生成可执行文件 .exe
# g++ [文件1.o] [文件2.o] ... -o [需要额外添加的动态库]
# 此处需要额外添加 -lthrift 的 thrift 动态库
g++ *.o -o main -lthrift

# 3. 运行 可执行文件
./main

将 非编译文件 和 非可执行文件 提交到 git 中去（好的工程习惯）

git add .
git restore --stage *.o     # .o文件是编译文件，不加入暂存区里
git restore --stage main    # main是可执行文件，不加入暂存区里
git commit -m "add match server"
git push

游戏应用端的客户端(match_client) #

同上，在 game 文件夹下实现 游戏应用端 的 客户端(match_client)

# 原始的thrift生成文件语法：thrift -r --gen <language> <Thrift filename>
mkdir game/src
cd game/src
thrift -r --gen py ../../thrift/match.thrift

# 将该文件夹重命名为 match_client
# 不过这里不改也无所谓，游戏应用端只有匹配的客户端
mv gen-py match_client

# 删掉 Match_remote 
# 该文件是用 py 实现 服务端 时用的文件
# 此处我们只需要实现 客户端 功能，因此他没有作用，不妨删掉，让文档简洁一点
rm match_client/match/Match-remote

利用官网提供的模板，编写客户端文件 client.py
官网模板[5]

删除前 4 行代码 用 Match 替代 Calculator 用 match_client. Match 替代 tutorial，修改成实际路径 删掉教学代码——transport.Open ()与 transport.Close ()之间的代码，替换成自己的业务代码 加入调试部分代码__main__ 一些课外知识Python实现Thrift Server[6]

client.py

 # 先删去开头前四行代码，该代码的作用只是将当前路径加入到环境变量里，我们用不到
 # 此处修改路径，以及Calculator 为 Match
 # 此处修改 ttypes 路径 以及 User 类
from match_client.match import Match  ## match.thrift 里实现的Service服务 Match
from match_client.match.ttypes import User ## match.thrift 里定义的struct结构User

from thrift import Thrift
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol


def main():
    # Make socket
    transport = TSocket.TSocket('localhost', 9090) # 客户端需要修改成服务端所在的IP地址

    # Buffering is critical. Raw sockets are very slow
    # 增加缓存区，提高socket速度
    transport = TTransport.TBufferedTransport(transport)

    # Wrap in a protocol 
    # 创建协议
    protocol = TBinaryProtocol.TBinaryProtocol(transport)

    # Create a client to use the protocol encoder 
    # 创建客户端
    client = Match.Client(protocol)  ## 从模板上粘贴下来时记得要修改客户端的名字，这里是Match

    # Connect!
    # 启动客户端
    transport.open()

    # 调试语句（其实是要在这里写具体的业务逻辑代码）
    user = User(1, 'timer', 1500)
    client.add_user(user, "") 

    # Close!
    transport.close()

 # 调用 main 函数
if __name__ == "__main__":
    main()

把文件保存到game/src下

要先开启匹配服务端(match_server)后运行client.py开启客户端(match_client)才不会报错！

acs@afcebfe3b1da:~/thrift_lesson/match_system/src$ ./main 
acs@afcebfe3b1da:~/thrift_lesson/game/src$ python3 client.py

如果服务端出现add_user，说明客户端创建成功。

这样我们就成功地用 python 调用了一个 C++ 代码了,实际上是 client.py 中调用了原 Match_server.skeleton.cpp (现 main.cpp)中 MatchHandler 类的 add_user 函数

将 非编译文件 和 非可执行文件 提交到 git 中去（好的工程习惯）

git add .
git restore --stage *.pyc   # .pyc文件是编译文件，不加入暂存区里
git restore --stage *.swp   # .swp文件是缓存文件，不加入暂存区里（vim没关时会生成）
git restore --stage "*.o"   # .o文件是编译文件，不加入暂存区里
git commit -m "add match_client"
git push

实现客户端业务逻辑 #

重写 client.py 使之能不断从终端里读入信息

client.py:

 # 利用 python 在终端读入信息需要引入 stdin
from sys import stdin

 # 将原来的通信 main 函数改写成operate函数，每次需要的时候调用一次建立通信传递信息
 # 目的是可以一直不断处理信息
 # 然后重写 main 函数，使之能不断从终端读入信息
def operate(op, user_id, user_name, score):
    # ...........................

    # 针对 op 参数，分别进行 "增加" 与 "删出" 操作
    user = User(user_id, user_name, score)

    if op == "add":
        client.add_user(user, "")
    else:
        client.remove_user(user, "")

    # ...........................

def main():
    for line in stdin:
        op, user_id, user_name, score = line.split(' ')
        operate(op, int(user_id), user_name, int(score))

此时运行客户端后，可在控制台读入数据 op user_id username score，例如 add 233 timer 1000

实现服务端业务逻辑 #

写server端需要分析业务模型。
需要进行监控(读用户进来)，支持增加用户和删除用户。同时还要有一个线程去不停地进行匹配，将信息传回给服务器端。
这个时候，我们发现需要实现两个功能，添加删除用户和匹配用户，根据微服务的概念需要开两个进程实现两个功能。

响应客户端请求和处理客户端请求可以拆分成两个独立过程，可使用多线程提高其效率。
假设使用两个线程完成服务端响应和处理过程：
一个线程负责响应客户端请求，接收客户端指令；
另一个线程负责处理指令，完成匹配。

这里需要运用到操作系统 里的 PV 原语 以及 生产者-消费者 模型

在本项目中，请求主要指客户端指令：添加用户add_user()和删除用户remove_user()，因此可用一个结构体Task描述该指令，其中type用于区分指令类型。

struct Task {
    User user;
    string type;        // "add"或"remove"
};

响应客户端请求的线程可看做生产者，它创建若干个Task对象；
处理客户端请求的线程可看做消费者，按照一定的规则删除 Task 对象，因此可用 生产者消费者模型 实现该过程。生产者消费者模型需要通信媒介，常用的一种实现方式是消费队列。

在代码实现中，消费队列是生产者进程和消费者进程的共享变量，多个线程同时修改它可能会导致结果出错，因此需要引入锁机制。当某个线程拿到消费队列的锁 mutex 后，消费队列只能由该线程使用，当另一个线程想使用时，会发现消费队列已上锁并进入阻塞状态，直到锁 mutex 被释放。

一些os概念：
互斥锁(S=1)：一个P操作(上锁)，一个V操作(解锁)
P操作的主要动作是:

  1. S 减1；
  2. 若 S 减1后仍大于或等于0，则进程继续执行；
  3. 若 S 减1后小于0，则该进程被阻塞后放入等待该信号量的 等待队列 中，然后转进程调度。
    V操作的主要动作是：
  4. S 加1；
  5. 若相加后结果大于0，则进程继续执行；
  6. 若相加后结果小于或等于0，则从该信号的等待队列中释放一个等待进程，然后再返回原进程继续执行或转进程调度。

对于P和V都是原子操作，就是在执行P和V操作时，不会被插队。从而实现对共享变量操作的原子性。
特殊:S=1表示互斥量，表示同一时间，信号量只能分配给一个线程。

多线程为啥要用锁? 因为多线程可能共享一个内存空间，导致出现重复读取并修改的现象。

消费者线程可看做是一个while(True) {...}的程序，它在main方法中创建。当没有task可消费时，会不停占用CPU资源，消耗系统资源，影响生产者线程接收数据。为了解决这个问题，可以使用条件变量condition_variable。条件变量可以让线程主动进入阻塞状态，直到被另一个线程的notify相关方法唤醒。

因此当消费者线程发现消费队列为空时，主动进入阻塞状态，**直到生产者进程接收客户端指令，修改消费者队列后再唤醒消费者线程消费task。**因此消费队列可按如下方式设计:

// 使用互斥锁的消费队列
struct MessageQueue {
    queue<Task> q;          // 消费队列
    mutex m;                // 互斥锁
    condition_variable cv;   // 条件变量，用于阻塞所在线程
}message_queue;

为了实现上述描述的生产者消费者模型，需要引入以下库文件。

#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <vector>

消费队列保存的是一个个待处理的任务task，而不是用户列表，因此需要创建一个类Pool，记录当前匹配池的情况以及定义匹配池的操作。匹配池Pool的操作主要包括：添加用户、删除用户、匹配、保存匹配结果。

其代码设计如下所示：

class Pool {
    // 匹配池
    public:
        void save_result(int a, int b) {
            printf("Match Result: %d %d\n", a, b);
        }

        void match() {
            while (users.size() > 1) {
                // 选择队头两名用户匹配
                auto a = users[0], b = users[1];
                users.erase(users.begin());
                users.erase(users.begin());

                save_result(a.id, b.id);
            }
        }

        void add(User user) {
            users.push_back(user);
        }

        void remove(User user) {
            // 根据id逐个查找用户，找到后删除
            for (uint32_t i = 0; i < users.size(); i++) // 用uint32_t变量防止size()出现warnning
                if (users[i].id == user.id) {
                    users.erase(users.begin() + i);
                    break;
                }
        }
    private:
        vector<User> users;         // 用户列表
}pool;

每次匹配选取用户列表里最靠前的两名用户来匹配，保存匹配记录后，移除这两名用户。

综上所述，可得到如下完整的服务端代码：

match_system/main.cpp

/ This autogenerated skeleton file illustrates how to build a server.
// You should copy it to another filename to avoid overwriting it.

#include "match_server/Match.h"
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>

#include <iostream>
#include <thread>               // 需要线程，引入头文件
#include <mutex>                // 互斥信号量
#include <condition_variable>   // 条件变量，用于 阻塞和唤醒 线程
#include <queue>                // 用于模拟消息队列
#include <vector>

using namespace ::apache::thrift;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::server;

using namespace  ::match_service;

using namespace std;


struct Task //消息队列中的元素
{
    User user;
    string type;
};

struct MessageQueue //消息队列
{
    queue<Task> q;          //消息队列本体
    mutex m;                //互斥信号量
    condition_variable cv;  //条件变量，用于阻塞唤醒线程
}message_queue;


class Pool  // 模拟匹配池
{
public:
    void save_result(int a, int b)  // 记录成功匹配的信息
    {
        printf("Match Result: %d %d\n", a, b);
    }

    void match()    // 匹配池中的第一、第二个用户进行匹配
    {
        while (users.size() > 1)
        {
            auto a = users[0], b = users[1];
            users.erase(users.begin());
            users.erase(users.begin());

            save_result(a.id, b.id);
        }
    }

    void add(User user) // 向匹配池中加入用户
    {
        users.push_back(user);
    }

    void remove(User user)  // 从匹配池中删除用户
    {
        for (uint32_t i = 0; i < users.size(); i ++ ) // 用uint32_t变量防止size()出现warnning
        {
            if (users[i].id == user.id)
            {
                users.erase(users.begin() + i);
                break;
            }
        }
    }

private:
    vector<User> users; // 匹配池中的用户，用 vector 记录
}pool;


class MatchHandler : virtual public MatchIf {
    public:
        MatchHandler() {
            // Your initialization goes here
        }

        int32_t add_user(const User& user, const std::string& info) {
            // Your implementation goes here
            printf("add_user\n");

            unique_lock<mutex> lck(message_queue.m);    // 访问临界区(消息队列)，先上锁（在函数里这样定义锁的时候会在函数结束时自动解锁，不需要再后面手动unlock）
            message_queue.q.push({user, "add"});        // 把新消息加入消息队列
            message_queue.cv.notify_all();              // 唤醒阻塞的线程

            return 0;
        }

        int32_t remove_user(const User& user, const std::string& info) {
            // Your implementation goes here
            printf("remove_user\n");

            unique_lock<mutex> lck(message_queue.m);    // 访问临界区(消息队列)，先上锁
            message_queue.q.push({user, "remove"});     // 把新消息加入消息队列
            message_queue.cv.notify_all();              // 唤醒阻塞的线程

            return 0;
        }

};


// 基于 "生产者-消费者模型" 的线程
void consume_task()
{
    while (true)
    {
        unique_lock<mutex> lck(message_queue.m);    // 访问临界区(消息队列)，先上锁

        if (message_queue.q.empty())
        {
            message_queue.cv.wait(lck);     //这里要阻塞进程,直到被其它线程的nofity方法唤醒
            // 避免队列为空时，一直反复运行该线程，导致一直占用临界区，而不能加入新消息
        }
        else
        {
            auto task = message_queue.q.front();    // 取出消息队列队头元素进行消费
            message_queue.q.pop();

            lck.unlock();   // 临界区访问结束，直接解锁(处理完共享变量后及时解锁)
            // 避免后续没用到临界区信息，而长时间占用临界区的情况发生

            if (task.type == "add") pool.add(task.user);
            else if (task.type == "remove") pool.remove(task.user);

            pool.match();
        }
    }
}


int main(int argc, char **argv) {
    int port = 9090;
    ::std::shared_ptr<MatchHandler> handler(new MatchHandler());
    ::std::shared_ptr<TProcessor> processor(new MatchProcessor(handler));
    ::std::shared_ptr<TServerTransport> serverTransport(new TServerSocket(port));
    ::std::shared_ptr<TTransportFactory> transportFactory(new TBufferedTransportFactory());
    ::std::shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());

    TSimpleServer server(processor, serverTransport, transportFactory, protocolFactory);

    cout << "Start Match Server" << endl;

    thread matching_thread(consume_task);   // 调用一个线程运行 consume_task

    server.serve();

    return 0;
}

由于服务端使用了线程库，因此在链接时，需要加参数-pthread链接线程相关库文件。
编译，运行，上传

g++ -c main.cpp
g++ *.o -o main -lthrift -pthread

git add main.cpp
git commit -m "match-server ver:2.0"  #（傻瓜匹配版）
git push

实现数据存储 (数据存储服务器与匹配系统服务器的交互) #

用thrift生成Save客户端cpp代码 (save.thrift)

save.thrift （现成）实现数据存储 (数据存储服务器 与 匹配系统服务器 的交互) 用thrift生成Save客户端cpp代码 (save.thrift)

save.thrift （现成）

namespace cpp save_service

service Save {

    /**
     * username: myserver的名称
     * password: myserver的密码的md5sum的前8位
     * 用户名密码验证成功会返回0，验证失败会返回1
     * 验证成功后，结果会被保存到myserver:homework/lesson_6/result.txt中
     */
    i32 save_data(1: string username, 2: string password, 3: i32 player1_id, 4: i32 player2_id)
    //acs_server的用户名，acs_server的密码的md5哈希值的前八位，匹配用户1，匹配用户2
}

在匹配系统服务器(match_server) 利用 thrift 生成 save.thrift 的 C++ 文件

并删掉 save.thrift 生成的 C++ 文件中不必要 服务端 文件(Save_server.skeleton.cpp)，因为在该交互功能里，数据存储服务器是作为客户端(save_client)的, 而且，一个服务器里只能有一个 main 函数执行, 只保留save_client里的 客户端 文件即可

不同的是，python 即使作为客户端，也可保留服务端的文件（如上文提到的 Match-Remote 文件），可删可不删，而 C++必须删掉服务端文件。因为一个节点(功能)只能由一个 main 方法作为程序的入口，所以匹配系统中的客户端和服务端写在同一个 main 方法中。我们这里根据逻辑将其实现在一个函数中。

cd match_system/src
thrift -r --gen cpp ../../thrift/save.thrift
mv gen-cpp save_client
rm save_client/Save_server.skeleton.cpp

将save_client端的代码写在match_server里
从 thrift官网[7] 复制 Client 端的模板到 main.cpp 下与 数据存储服务器 交互的函数中，然后开始修改模板里的一些参数。

match_system/src的main.cpp

① 加入模板中需要，但 main.cpp 没有的头文件

#include <thrift/transport/TTransportUtils.h>
#include <thrift/transport/TSocket.h>

② 引入生成的Save.h

#include "save_client/Save.h"

③添加save.thrift定义的命名空间，保证代码正确引用Save.h的内容

using namespace ::save_service;

④ 把模板main方法里的内容拷贝到match_system/src/main.cpp的Pool类的void save_result(int a, int b)方法的printf(...)后边，并用gg=G格式化代码，然后按如下修改

  • 把粘贴代码里new TSocket("localhost", 9090)的localhost改成第4讲配置的myserver的IP
  • 把CalculatorClient换成SaveClient
  • 删除try语句块里的transport->open();与transport->close();之间的教学语句，然后加入语句client.save_data("myserver_username", "密码md5前八位", a, b);，注意
    myserver_username指第4章配置的myserver的用户名，可通过homework 4 getinfo查看 为了防止密码泄露风险，校验采用密码 md5 码的前八位，可通过命令 echo your_password | md5sum | cut -c 1-8 获得，其中 your_password 是你的明文密码
std::shared_ptr<TTransport> socket(new TSocket("xx.xx.xx.xx", 9090));       // xx.xx.xx.xx为自己myserver的IP
std::shared_ptr<TTransport> transport(new TBufferedTransport(socket));
std::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
SaveClient client(protocol);

try {
    transport->open();

    client.save_data("acs_1234", "abcdefgh", a, b);     // 替换换成自己myserver的用户名和密码mk5前八位

    transport->close();
} catch (TException& tx) {
    cout << "ERROR: " << tx.what() << endl;
}

修改后的main.cpp代码如下：

#include "match_server/Match.h"
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>
// 需要额外引入的头文件
#include <thrift/transport/TTransportUtils.h>
#include <thrift/transport/TSocket.h>
#include "save_client/Save.h"
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <vector>

using namespace ::apache::thrift;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::server;

using namespace ::match_service;
// 需要额外声明的命名空间
using namespace ::save_service;
using namespace std;

struct Task {
    User user;
    string type;
};

struct MessageQueue {
    queue<Task> q;
    mutex m;
    condition_variable cv;
}message_queue;

class Pool {
    public:
    //重写 save_result 内的内容，使其能够与 "数据存储服务器" 交互
        void save_result(int a, int b) { // 记录成功匹配的信息
            printf("Match Result: %d %d\n", a, b);
            // Client端的板子
            std::shared_ptr<TTransport> socket(new TSocket("123.57.47.211", 9090));
            std::shared_ptr<TTransport> transport(new TBufferedTransport(socket));
            std::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
            SaveClient client(protocol);

            try {
                transport->open();

                //调用接口，把信息存储 "数据存储服务器" 中
                int res = client.save_data("acs_2", "50c0f4b2", a, b);
                //输出匹配结果
                if (!res) puts("success");
                else puts("fail");

                transport->close();
            } catch (TException& tx) {
                cout << "ERROR: " << tx.what() << endl;
            }
        }

        void match() {
            while (users.size() > 1) {
                auto a = users[0], b = users[1];
                users.erase(users.begin());
                users.erase(users.begin());

                save_result(a.id, b.id);
            }
        }

        void add(User user) {
            users.push_back(user);
        }

        void remove(User user) {
            for (uint32_t i = 0; i < users.size(); i++)
                    if (users[i].id == user.id) {
                            users.erase(users.begin() + i);
                            break;
                    }
        }
    private:
        vector<User> users;

}pool;

class MatchHandler : virtual public MatchIf {
    public:
        MatchHandler() {
            // Your initialization goes here
        }

        /**
         * user: 添加的用户信息
         * info: 附加信息
         * 在匹配池中添加一个名用户
         * 
         * @param user
         * @param info
         */
        int32_t add_user(const User& user, const std::string& info) {
            // Your implementation goes here
            printf("add_user\n");
            unique_lock<mutex> lck(message_queue.m);
            message_queue.q.push({user, "add"});
            message_queue.cv.notify_all();

            return 0;
        }

        /**
         * user: 删除的用户信息
         * info: 附加信息
         * 从匹配池中删除一名用户
         * 
         * @param user
         * @param info
         */
        int32_t remove_user(const User& user, const std::string& info) {
            // Your implementation goes here
            printf("remove_user\n");

            unique_lock<mutex> lck(message_queue.m);
            message_queue.q.push({user, "remove"});
            message_queue.cv.notify_all();
            return 0;
        }

};

void consume_task() {
    while (true) {
        unique_lock<mutex> lck(message_queue.m);
        if (message_queue.q.empty()) {
            message_queue.cv.wait(lck);
        } else {
            auto task = message_queue.q.front();
            message_queue.q.pop();
            lck.unlock();

            if (task.type == "add") pool.add(task.user);
            else if (task.type == "remove") pool.remove(task.user);

            pool.match();
        }
    }
}

int main(int argc, char **argv) {
    int port = 9090;
    ::std::shared_ptr<MatchHandler> handler(new MatchHandler());
    ::std::shared_ptr<TProcessor> processor(new MatchProcessor(handler));
    ::std::shared_ptr<TServerTransport> serverTransport(new TServerSocket(port));
    ::std::shared_ptr<TTransportFactory> transportFactory(new TBufferedTransportFactory());
    ::std::shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());

    TSimpleServer server(processor, serverTransport, transportFactory, protocolFactory);
    std::cout << "Start Match Server" << std::endl;

    std::thread matching_thread(consume_task);

    server.serve();
    return 0;
}

⑤ 编译及链接代码

g++ -c main.cpp save_client/*.cpp match_server/*.cpp
g++ *.o -o main -lthrift -pthread

⑥ 检验

首先把game/src/client.py的operate方法的localhost改成127.0.0.1

然后在tmux开启两个bash，分别在match_system/src执行./main 和 game/src执行python3 client.py

在客户端输入若干指令，观察服务端的匹配情况。

最后登录保存数据的服务器 ssh myserver，查看 ~/homework/lesson_6/result.txt 是否存在，是否有匹配信息。

-------------
# 游戏应用端
add 1 yxc 1500
add 2 zxc 1500
# 匹配系统服务器
add_user
add_user
Match Result: 1 2
success
# 数据存储服务器
## homework/lesson_6/result.txt
result: 1 2
141da30f
-------------

# 上传到 git 服务器
git add main.cpp
git add ../../thrift/save.thrift
git commit -m "implement save-client"
git push

实现匹配系统 ver3.0 功能：每次只匹配分差小于50的用户 #

编写匹配逻辑，修改 match_system/src/main.cpp

修改线程中 消息队列 为空时，不再是 阻塞 直到 唤醒 为止

让其可以每经过 1 秒就进行一次 match() 调用

这样做的目的是：由于我们的匹配策略发生了变化(即每次只匹配分差小于50的用户)，可能匹配池中仍然有用户在等待匹配（当前匹配池各个用户分差都大于50），而消息队列此时仍为空

如果仍然采用先前的策略，可能会导致进程卡死（新用户不进去，老用户永远匹配不了）

如果不加入 “随时间扩大匹配域” 的功能，无论改不改上述情况都会可能发生 但是为了后续实现 “随时间扩大匹配域” 功能，先做一个铺垫

（1）改成每1秒匹配一次

① 去掉消费者进程方法consume_task的阻塞代码，让它解锁后直接匹配，然后休眠1s

...
//修改 "生产者 - 消费者模型" 的线程中，关于消息队列为空时的处理
if (message_queue.q.empty())
{
    // 此处修改为每 1 秒进行一次匹配，而不是等到被唤醒时才匹配
    lck.unlock();   // 直接解锁临界区资源
    pool.match;     //调用math()
}
...

② 引入sleep()需要的头文件

#include <unistd.h>

（2）编写匹配逻辑

首先按分值升序排序，依次检查相邻用户的分值差的绝对值是否小于50，如果满足立即匹配这两名用户。

修改Pool类中match()方法的匹配逻辑:

//重写匹配池Pool类中的match函数，使之可以匹配分数差在 50 以内的两个用户
void match()    // 匹配池中的第一、第二个用户进行匹配
{
    while (users.size() > 1)
    {
        // 按照 rank分 排序
        sort(users.begin(), users.end(), [&](User &a, User &b){
            return a.score < b.score;
        });
        bool flag = true;
        for (uint32_t i = 1; i < users.size(); i ++ )
        {
            User a = users[i - 1], b = users[i];
            // 两名玩家分数差小于50时进行匹配
            if (b.score - a.score <= 50)
            {
                users.erase(users.begin() + i - 1, users.begin() + i); //删掉用户a,b
                save_result(a.id, b.id);

                flag = false;
                break;
            }
        }
        if (flag) break;    // 一轮扫描后，发现没有能够匹配的用户，就停止扫描，等待下次调用
    }
}

（3）验证匹配逻辑

g++ -c main.cpp
g++ *.o -o main -lthrift -pthread
./main

-------------
# 游戏应用端
add 1 yxc 2000
add 2 yume 1500
add 3 minami 1549
# 匹配系统服务器
add_user
add_user    # 此处用户 yxc 与用户 yume 没有匹配，因为分差超过50
add_user
Match Result: 2 3   # 用户 yume 与用户 minami 分差在 50 以内，进行匹配
success
# 数据存储服务器
## homework/lesson_6/result.txt
result: 2 3
b6281235
-------------

# 上传到 git 服务器
git add main.cpp
git add ../../thrift/save.thrift
git commit -m "match server:3.0"
git push

实现匹配系统 ver4.0 功能：多线程并发 #

我们之前的是TSimpleServer是单线程的，效率比较慢，我们接下来将它升级为多线程TThreadedServer
参考官网 C++服务端[8] 模板，修改match_system/src/main.cpp

（1）添加缺少的头文件 #
#include <thrift/concurrency/ThreadManager.h>
#include <thrift/concurrency/ThreadFactory.h>
#include <thrift/server/TThreadedServer.h>
#include <thrift/TToString.h>
（2）替换掉main()方法里的服务器构建过程 #
TThreadedServer server(
    std::make_shared<CalculatorProcessorFactory>(std::make_shared<CalculatorCloneFactory>()),
    std::make_shared<TServerSocket>(9090), //port
    std::make_shared<TBufferedTransportFactory>(),
    std::make_shared<TBinaryProtocolFactory>()
  );
（3）复制工厂代码到main方法上边，注释掉输出信息 #
/*
  CalculatorIfFactory is code generated.
  CalculatorCloneFactory is useful for getting access to the server side of the
  transport.  It is also useful for making per-connection state.  Without this
  CloneFactory, all connections will end up sharing the same handler instance.
*/
class CalculatorCloneFactory : virtual public CalculatorIfFactory {
 public:
  ~CalculatorCloneFactory() override = default;
  CalculatorIf* getHandler(const ::apache::thrift::TConnectionInfo& connInfo) override
  {
    std::shared_ptr<TSocket> sock = std::dynamic_pointer_cast<TSocket>(connInfo.transport);
    // cout << "Incoming connection\n";
    // cout << "\tSocketInfo: "  << sock->getSocketInfo() << "\n";
    // cout << "\tPeerHost: "    << sock->getPeerHost() << "\n";
    // cout << "\tPeerAddress: " << sock->getPeerAddress() << "\n";
    // cout << "\tPeerPort: "    << sock->getPeerPort() << "\n";
    return new CalculatorHandler;
  }
  void releaseHandler( ::shared::SharedServiceIf* handler) override {
    delete handler;
  }
};

将Calculator替换成Match命令：

:1,$s/Calculator/Match/g
（4）修改参数releaseHandler #
void releaseHandler( MatchIf* handler) override {
    delete handler;
}

整合一下得到：

// 引入新的多线程头文件
#include <thrift/concurrency/ThreadManager.h>
#include <thrift/concurrency/ThreadFactory.h>
#include <thrift/server/TThreadPoolServer.h>
#include <thrift/server/TThreadedServer.h>
#include <thrift/TToString.h>

// 复制模板的类 CalculatorCloneFactory 然后改一改，把所有的 Calculator 改为 Match
class MatchCloneFactory : virtual public MatchIfFactory {
    public:
        ~MatchCloneFactory() override = default;
        MatchIf* getHandler(const ::apache::thrift::TConnectionInfo& connInfo) override
        {
            std::shared_ptr<TSocket> sock = std::dynamic_pointer_cast<TSocket>(connInfo.transport);
            /*
            cout << "Incoming connection\n";
            cout << "\tSocketInfo: "  << sock->getSocketInfo() << "\n";
            cout << "\tPeerHost: "    << sock->getPeerHost() << "\n";
            cout << "\tPeerAddress: " << sock->getPeerAddress() << "\n";
            cout << "\tPeerPort: "    << sock->getPeerPort() << "\n";
            */
            return new MatchHandler;
        }
        void releaseHandler(MatchIf* handler) override {    //改为MatchIf*  
            delete handler;
        }
};
// 重写main函数，启用多线程服务器
int main(int argc, char **argv) {
    TThreadedServer server(
            std::make_shared<MatchProcessorFactory>(std::make_shared<MatchCloneFactory>()),
            std::make_shared<TServerSocket>(9090), //port
            std::make_shared<TBufferedTransportFactory>(),
            std::make_shared<TBinaryProtocolFactory>());

    cout << "Start Match Server" << endl;

    thread matching_thread(consume_task);   // 调用一个线程运行 consume_task

    server.serve();

    return 0;
}

实现匹配系统 ver5.0功能：随时间扩大匹配域 #

思想 #

如果匹配池有两个人不满足分值差不超过50，按之前的逻辑这两个人永远不会被匹配，但这样体验不好，因此引入动态匹配。每个人允许的分值差是动态变化的，它等于等待时间∗50等待时间∗50。如果两个人的分值差都在各自允许的分值差范围内，则匹配这两人。

例如甲的分值为1000分，乙为1500分。甲已经等待了11秒，其允许分值差为550分；乙等待了9秒，其运输分值差为450分。尽管二者分数差在甲当前的容忍范围内，但不在乙的容忍范围内，因此不匹配。再过1秒后，时间差也在乙的容忍范围内了，可以匹配了。

实现 #

（1）修改match_system/src/main.cpp的Pool类

① 引入等待时间成员变量vector<int> wt;

② 让add_user()和remove_user()支持wt的添加与删除

③ 修改match()方法并引入check_match()方法

得到的Pool类如下:

class Pool {
    public:
        void save_result(int a, int b) {
            printf("Match Result: %d %d\n", a, b);

            std::shared_ptr<TTransport> socket(new TSocket("123.57.47.211", 9090));
            std::shared_ptr<TTransport> transport(new TBufferedTransport(socket));
            std::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
            SaveClient client(protocol);

            try {
                transport->open();

                client.save_data("acs_3929", "6df6b19d", a, b);

                transport->close();
            } catch (TException& tx) {
                cout << "ERROR: " << tx.what() << endl;
            }
        }

        bool check_match(uint32_t i, uint32_t j) 
        {
            auto a = users[i], b = users[j]; 
            int dt = abs(a.score - b.score);
            int a_max_dif = wt[i] * 50;
            int b_max_dif = wt[j] * 50;

            return dt <= a_max_dif && dt <= b_max_dif;
        }

        void match()
        {
            for (uint32_t i = 0; i < wt.size(); i++)
                wt[i]++;                // 更新等待时间

            while(users.size() > 1)
            {
                bool flag = true;
                for (uint32_t i = 0; i < users.size(); i++)
                {
                    for (uint32_t j = i + 1; j < users.size(); j++) 
                    {
                        if (check_match(i, j))
                        {
                            auto a = users[i], b = users[j];
                            users.erase(users.begin() + j);             // 使用erase删除时，要先删后边的
                            wt.erase(wt.begin() + j);
                            users.erase(users.begin() + i);             // 再删前边的
                            wt.erase(wt.begin() + i);        
                            save_result(a.id, b.id);
                            flag = false;
                            break;
                        }
                    }
                }
                if (flag) break;                // 匹配成功后立即停止，防止进入死循环
            }
        }

        void add(User user) {
            users.push_back(user);
            wt.push_back(0);
        }

        void remove(User user) {
            for (uint32_t i = 0; i < users.size(); i++)
                if (users[i].id == user.id) {
                    users.erase(users.begin() + i);
                    wt.erase(wt.begin() + i);
                    break;
                }
        }
    private:
        vector<User> users;
        vector<int> wt;  // 等待时间

}pool;

（2）删除consume_task()里else里的pool.match()，保证先匹配，且等待时间正确。

修改完后的match_server的main.cpp:

#include "match_server/Match.h"
#include "save_client/Save.h"

#include <thrift/concurrency/ThreadManager.h>
#include <thrift/concurrency/ThreadFactory.h>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/server/TThreadPoolServer.h>
#include <thrift/server/TThreadedServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportUtils.h>
#include <thrift/transport/TBufferTransports.h>
#include <thrift/TToString.h>

#include <iostream>
#include <thread>               // 需要线程，引入头文件
#include <mutex>                // 互斥信号量
#include <condition_variable>   // 条件变量，用于 阻塞和唤醒 线程
#include <queue>                // 用于模拟消息队列
#include <vector>
#include <unistd.h>             // 用于调用 sleep 函数

using namespace ::apache::thrift;
using namespace ::apache::thrift::concurrency;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::server;

using namespace  ::match_service;
using namespace  ::save_service;

using namespace std;


struct Task //消息队列中的元素
{
    User user;
    string type;
};

struct MessageQueue //消息队列
{
    queue<Task> q;          //消息队列本体
    mutex m;                //互斥信号量
    condition_variable cv;  //条件变量，用于阻塞唤醒线程
}message_queue;


class Pool  // 模拟匹配池
{
    public:
        void save_result(int a, int b)  // 记录成功匹配的信息
        {
            printf("Match Result: %d %d\n", a, b);

            // Client端的板子
            std::shared_ptr<TTransport> socket(new TSocket("123.57.47.211", 9090));
            std::shared_ptr<TTransport> transport(new TBufferedTransport(socket));
            std::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
            SaveClient client(protocol);

            try {
                transport->open();

                //调用接口，把信息存储 "数据存储服务器" 中
                int res = client.save_data("acs_2", "50c0f4b2", a, b);
                if (!res) puts("success");
                else puts("fail");

                transport->close();
            } catch (TException& tx) {
                cout << "ERROR: " << tx.what() << endl;
            }
        }

        bool check_match(uint32_t i,uint32_t j)
        {
            auto a = users[i], b = users[j];

            int dt = abs(a.score - b.score);
            int a_max_dif = wt[i] * 50;
            int b_max_dif = wt[j] * 50;

            return dt <= a_max_dif && dt <= b_max_dif;


        }

        void match()    // 匹配池中的第一、第二个用户进行匹配
        {
            //每次匹配的时候，每个玩家等待时间加一
            for (uint32_t i = 0; i < wt.size(); i ++ )
            {
                wt[i] ++; //等待秒数加一
            }

            while (users.size() > 1)
            {
                bool flag = true;
                for (uint32_t i = 0; i < users.size(); i ++ )
                {
                    for (uint32_t j = i + 1;j < users.size(); j ++)
                    {
                        if (check_match(i,j))
                        {
                            //先删后面的再删前面的，不然会影响下标
                            auto a = users[i], b = users[j];
                            users.erase(users.begin() + j);
                            users.erase(users.begin() + i);
                            wt.erase(wt.begin() + j);
                            wt.erase(wt.begin() + i);
                            save_result(a.id,b.id);
                            flag = false;
                            break;
                        }
                    }
                    if (!flag) break;
                }
                if (flag) break;    // 一轮扫描后，发现没有能够匹配的用户，就停止扫描，等待下次调用
            }
        }

        void add(User user) // 向匹配池中加入用户
        {
            users.push_back(user);
            wt.push_back(0);
        }

        void remove(User user)  // 从匹配池中删除用户
        {
            for (uint32_t i = 0; i < users.size(); i ++ )
            {
                if (users[i].id == user.id)
                {
                    users.erase(users.begin() + i);
                    wt.erase(wt.begin() + i);
                    break;
                }
            }
        }

    private:
        vector<User> users; // 匹配池中的用户，用 vector 记录
        vector<int> wt; // 等待时间，单位：s
}pool;


class MatchHandler : virtual public MatchIf {
    public:
        MatchHandler() {
            // Your initialization goes here
        }

        int32_t add_user(const User& user, const std::string& info) {
            // Your implementation goes here
            printf("add_user\n");

            unique_lock<mutex> lck(message_queue.m);    // 访问临界区(消息队列)，先上锁
            message_queue.q.push({user, "add"});        // 把新消息加入消息队列
            message_queue.cv.notify_all();              // 唤醒阻塞的线程

            return 0;
        }

        int32_t remove_user(const User& user, const std::string& info) {
            // Your implementation goes here
            printf("remove_user\n");

            unique_lock<mutex> lck(message_queue.m);    // 访问临界区(消息队列)，先上锁
            message_queue.q.push({user, "remove"});     // 把新消息加入消息队列
            message_queue.cv.notify_all();              // 唤醒阻塞的线程

            return 0;
        }

};


// 基于 "生产者-消费者模型" 的线程
void consume_task()
{
    while (true)
    {
        unique_lock<mutex> lck(message_queue.m);    // 访问临界区(消息队列)，先上锁

        if (message_queue.q.empty())
        {
            // 此处修改为每 1 秒进行一次匹配，而不是等到被唤醒时才匹配
            lck.unlock();   // 直接解锁临界区资源，启用每秒调用的策略
            pool.match();   // 调用 match()
            sleep(1);       // 休眠 1 秒
        }
        else
        {
            auto task = message_queue.q.front();    // 取出消息队列队头元素
            message_queue.q.pop();

            lck.unlock();   // 临界区访问结束，直接解锁
            // 避免后续没用到临界区信息，而长时间占用临界区的情况发生

            if (task.type == "add") pool.add(task.user);
            else if (task.type == "remove") pool.remove(task.user);
        }
    }
}
// 复制模板的类 CalculatorCloneFactory 然后改一改，把所有的 Calculator 改为 Match
class MatchCloneFactory : virtual public MatchIfFactory {
    public:
        ~MatchCloneFactory() override = default;
        MatchIf* getHandler(const ::apache::thrift::TConnectionInfo& connInfo) override
        {
            std::shared_ptr<TSocket> sock = std::dynamic_pointer_cast<TSocket>(connInfo.transport);
            /*
            cout << "Incoming connection\n";
            cout << "\tSocketInfo: "  << sock->getSocketInfo() << "\n";
            cout << "\tPeerHost: "    << sock->getPeerHost() << "\n";
            cout << "\tPeerAddress: " << sock->getPeerAddress() << "\n";
            cout << "\tPeerPort: "    << sock->getPeerPort() << "\n";
            */
            return new MatchHandler;
        }
        void releaseHandler(MatchIf* handler) override {    //改为MatchIf*
            delete handler;
        }
};

int main(int argc, char **argv) {
    TThreadedServer server(
            std::make_shared<MatchProcessorFactory>(std::make_shared<MatchCloneFactory>()),
            std::make_shared<TServerSocket>(9090), //port
            std::make_shared<TBufferedTransportFactory>(),
            std::make_shared<TBinaryProtocolFactory>());

    cout << "Start Match Server" << endl;

    thread matching_thread(consume_task);   // 调用一个线程运行 consume_task

    server.serve();

    return 0;
}

编译，运行, 验证，上传

g++ -c main.cpp
g++ *.o -o main -lthrift -pthread
./main

-------------
# 游戏应用端
add 1 yxc 2000
add 2 zxc 1500
# 匹配系统服务器
add_user
add_user
# 等待了 10 s
Match Result: 1 2
success
# 数据存储服务器
## homework/lesson_6/result.txt
result: 1 2
141da30f
-------------

# 上传到 git 服务器
git add main.cpp
git commit -m "match server:5.0"
git push

localhost，127.0.0.1 和本机 IP 三者的区别 #

localhost 是本机访问； 127.0.0.1 是本机访问；本机 IP 是本机或外部访问。 localhost 是域名，默认是指向 127.0.0.1 ；而本机 IP 就是本机对外放开访问的 IP 地址，这个网址就是与物理网卡绑定的 IP 地址。

如果把IP地址比作一间房子 ，端口就是出入这间房子的门。真正的房子只有几个门，但是一个IP地址的端口可以有65536（即：2^16）个之多！端口是通过端口号来标记的，端口号只有整数，范围是从0 到65535（2^16-1）。 同一个端口只能由一个进程来监听。所以我们一旦启动了一个服务，那么这个服务就不能在被另一个进程启动了。 服务器的端口号要与客户端的端口号相同。

每执行一个程序就是开了一个进程。每一个进程可以开一堆线程。开多线程的开销是很小的，开多进程的开销是很大的。

更多参考： https://git.acwing.com/fashen/thrift_learning[9]

References


Related readings

<< prev | LaTex basic... Continue strolling Building a... ｜ next >>
If you want to follow my updates, or have a coffee chat with me, feel free to connect with me:
Follow @timerring @timerringX 知乎@timerring Official Account

© 2022 - present timerring