🦈Thrift

• Thrift | AcWing Linux 基础课

• Thrift_learning - AcGit

• AcWing Linux 基础课 | Colopen's blog | Thrift

• thrift 官网 -> Tutorial -> tutorial.thrift

---

Thrift是什么？

Thrift是一个轻量级、跨语言的远程过程服务调用（RPC）框架

RPC（远程过程调用）是一个计算机通信协议，该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程

Thrift用于跨语言服务开发，它将软件栈和代码生成引擎结合在一起，以构建在 C++、Java、Python、PHP、Ruby、Erlang、Perl、Haskell、C#、Cocoa、JavaScript、Node. Js、Smalltalk、OCaml 和 Delphi 等语言之间高效、无缝地工作的服务

例：实现一个游戏的匹配服务

基本框架

1. 游戏应用端 game（Python3）

   1. 客户端：与 匹配系统服务器 的服务端交互

2. 匹配系统服务器 match_system（C++）

   1. 服务端：与 游戏应用端 的客户端交互

   2. 客户端：与 数据存储服务器 的服务端交互

3. 数据存储服务器（已经实现）

   1. 服务端：与 匹配系统服务器 的客户端交互

文件结构

|-- README.md
|-- game
|   `-- src
|       |-- client.py
|       `-- match_client
|           |-- __init__.py
|           |-- __pycache__
|           |   `-- __init__.cpython-38.pyc
|           `-- match
|               |-- Match.py
|               |-- __init__.py
|               |-- __pycache__
|               |   |-- Match.cpython-38.pyc
|               |   |-- __init__.cpython-38.pyc
|               |   `-- ttypes.cpython-38.pyc
|               |-- constants.py
|               `-- ttypes.py
|-- match_system
|   `-- src
|       |-- Match.o
|       |-- Save.o
|       |-- main
|       |-- main.cpp
|       |-- main.o
|       |-- match_server
|       |   |-- Match.cpp
|       |   |-- Match.h
|       |   |-- match_types.cpp
|       |   `-- match_types.h
|       |-- match_types.o
|       `-- save_client
|           |-- Save.cpp
|           |-- Save.h
|           `-- save_types.h
`-- thrift
    |-- match.thrift
    `-- save.thrift

实现过程

1. 定义接口 (thrift 文件夹用于存放接口 )

2. 完成 Server 端

   1. 通过match.thrift接口在match_system文件夹下生成 C++版本的服务端

   thrift -r --gen cpp tutorial.thrift

   1. 将gen cpp 文件夹重命名，如：match_server（区别于之后要在此处生成的client_server）

   2. 把 Match_server.skeleton.cpp 移动到当前 src 目录下并重命名为 main.cpp

      • 由于移动了 main.cpp 故需要修改一下 main.cpp 中头文件里关于 Match.h 的引用路径：#include "Match.h" -> #include "match_server/Match.h"

   3. 在main.cpp中实现具体业务逻辑

3. 完成 Client 端

   1. 通过match.thrift接口在game文件夹下生成 python3 版本的服务端，然后通过修改得到客户端

   thrift -r --gen py tutorial.thrift

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

   2. 利用官网提供的模板，在src文件夹下编写 客户端 文件 client.py

4. 持久化到云端

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

     • Cpp

       • git restore --stage *.o

       • git restore --stage main

     • Python

       • git restore --stage *.pyc # .pyc文件是编译文件，不加入暂存区里

       • git restore --stage *.swp # .swp文件是缓存文件，不加入暂存区里

• 注意：先运行服务器后，客户端才能正常运行

Thrift 接口

Match.thrift

namespace cpp match_service
struct User {
    1: i32 id,
    2: string name,
    3: i32 scores
}
service Match {
    i32 add_user(1: User user, 2: string info),
    i32 remove_user(1: User user, 2: string info),
}

Save.thrift

namespace cpp save_service
service Save {
    /**
     * username: myserver的名称
     * password: myserver的密码的md5值的前8位，用命令md5sum
     * 用户名密码验证成功会返回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)
}

各版本预览

Match_server:1.0

• match_client：创建固定的User

from match_client.match import Match
from match_client.match.ttypes import 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)

    # Buffering is critical. Raw sockets are very slow
    transport = TTransport.TBufferedTransport(transport)

    # Wrap in a protocol
    protocol = TBinaryProtocol.TBinaryProtocol(transport)

    # Create a client to use the protocol encoder
    client = Match.Client(protocol)

    # Connect!
    transport.open()

    user = User(1, 'yxc', 1500)
    client.add_user(user, "")

    # Close!
    transport.close()


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

• match_server

// 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>
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;
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");
    return 0;
  }
  int32_t remove_user(const User& user, const std::string& info) {
    // Your implementation goes here
    printf("remove_user\n");
    return 0;
  }
};
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;
    
  server.serve();
  return 0;
}

Match_server:2.0

• match_client：根据标准输入来创建User

from match_client.match import Match
from match_client.match.ttypes import User

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

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

# 将原来的通信 main 函数改写成operate函数，每次需要的时候调用一次建立通信传递信息
# 目的是可以一直不断处理信息
# 然后重写 main 函数，使之能不断从终端读入信息
def operate(op, user_id, user_name, score):
    # Make socket 
	transport = TSocket.TSocket('localhost', 9090)
	
	# Buffering is critical. Raw sockets are very slow
	transport = TTransport.TBufferedTransport(transport)
	
	# Wrap in a protocol
	protocol = TBinaryProtocol.TBinaryProtocol(transport)
	
	# Create a client to use the protocol encoder
	client = Match.Client(protocol)
	
	# Connect!
	transport.open()

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

    if op == "add":
        client.add_user(user, "")
    else:
        client.remove_user(user, "")
    
    # Close!
	transport.close()

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

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

• match_server：自动将用户池中前两个用户匹配到一起

// 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) {
            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);    // 访问临界区（消息队列），先上锁
    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); // 这里要阻塞进程
            // 避免队列为空时，一直反复运行该进程，导致一直占用临界区，而不能加入新消息
        } 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;
}

Match_server:3.0

• save_client：因为一个节点只能由一个main方法作为程序的入口，所以匹配系统中的客户端和服务端写在同一个main方法中

// 需要额外引入的头文件
#include "save_client/Save.h"
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportUtils.h>

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

//重写 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_4888", "07637c4c", a, b);
        //输出匹配结果
        if (!res) puts("success");
        else puts("fail");

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

• match_server：每次只匹配分差小于 50 的用户

// This autogenerated skeleton file illustrates how to build a server.
// You should copy it to another filename to avoid overwriting it.
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportUtils.h>
#include "match_server/Match.h"
#include "save_client/Save.h"
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>
#include <unistd.h> // 用于调用 sleep 函数
#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 ::save_service;
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:
	//重写 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_4888", "07637c4c", 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) {
            // 按照 rank分 排序
            sort(users.begin(), users.end(), [&](User& a, User& b) {
                return a.scores < b.scores;
            });
            bool flag = true;
            for (uint32_t i = 1; i < users.size(); ++ i) {
                auto a = users[i - 1], b = users[i];
                // 两名玩家分数差小于50时进行匹配
                if (b.scores - a.scores <= 50) {
                    users.erase(users.begin() + i - 1, users.begin() + i + 1);
                    save_result(a.id, b.id);
                    flag = false;
                    break;
                }
            }
            if (flag) break;    // 一轮扫描后，发现没有能够匹配的用户，就停止扫描，等待下次调用
        }
    } 
    
    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; // 匹配池中的用户，用 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);    // 访问临界区（消息队列），先上锁
    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); // 这里要阻塞进程
            // 避免队列为空时，一直反复运行该进程，导致一直占用临界区，而不能加入新消息
            // 修改为每 1 秒进行一次匹配
            lck.unlock();   // 直接解锁临界区资源
            pool.match();   //调用 match()
            sleep(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);
            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;
}

Match_server:4.0

• match_server：随时间扩大匹配域，每一单位的 wt 会扩大 $50$ 分 的匹配域

// This autogenerated skeleton file illustrates how to build a server.
// You should copy it to another filename to avoid overwriting it.
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportUtils.h>
#include "match_server/Match.h"
#include "save_client/Save.h"
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>
#include <unistd.h> // 用于调用 sleep 函数
#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 ::save_service;
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);
        // 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_4888", "07637c4c", 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.scores - b.scores);
        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()) {
            // message_queue.cv.wait(lck); // 这里要阻塞进程
            // 避免队列为空时，一直反复运行该进程，导致一直占用临界区，而不能加入新消息
            // 修改为每 1 秒进行一次匹配
            lck.unlock();   // 直接解锁临界区资源
            pool.match();   //调用 match()
            sleep(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);
        } 
    }
}
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;
}

Usage

# 启动服务端
./match_system/src/main

# 启动客户端
python3 game/src/client.py

# 游戏应用端 （op id name scores）
add 1 yxc 2000
add 2 xan 1500
add 3 zagy 2500
remove 3 zagy 2500

# 匹配系统服务器
add_user
add_user
remove_user 
# 等待了 10 s
Match Result: 1 2
success

项目地址

• Learning-Thrift