1. Lars-reportV0.1 report模块介绍

5) 存储线程池及消息队列

        我们现在的reporter_service的io入库操作，完全是在消息的callback中进行的，那么实际上，这回占用我们server的工作线程的阻塞时间，从而浪费cpu。所以我们应该将io的入库操作，交给一个专门做入库的消息队列线程池来做，这样我们的callback就会立刻返回该业务，从而可以继续处理下一个conn链接的消息事件业务。

        所以我们就要在此给reporter_service设计一个存储数据的线程池及配套的消息队列。当然这里面我们还是直接用写好的`lars_reactor`框架里的接口即可。

> lars_reporter/src/reporter_service.cpp

```c
#include "lars_reactor.h"
#include "lars.pb.h"
#include "store_report.h"
#include <string>

thread_queue<lars::ReportStatusRequest> **reportQueues = NULL;
int thread_cnt = 0;

void get_report_status(const char *data, uint32_t len, int msgid, net_connection *conn, void *user_data)
{
    lars::ReportStatusRequest req;

    req.ParseFromArray(data, len);

    //将上报数据存储到db
    StoreReport sr;
    sr.store(req);

    //轮询将消息平均发送到每个线程的消息队列中
    static int index = 0;
    //将消息发送给某个线程消息队列
    reportQueues[index]->send(req);
    index ++;
    index = index % thread_cnt;
}

void create_reportdb_threads()
{
    thread_cnt = config_file::instance()->GetNumber("reporter", "db_thread_cnt", 3);

    //开线程池的消息队列
    reportQueues = new thread_queue<lars::ReportStatusRequest>*[thread_cnt];

    if (reportQueues == NULL) {
        fprintf(stderr, "create thread_queue<lars::ReportStatusRequest>*[%d], error", thread_cnt) ;
        exit(1);
    }

    for (int i = 0; i < thread_cnt; i++) {
        //给当前线程创建一个消息队列queue
        reportQueues[i] = new thread_queue<lars::ReportStatusRequest>();
        if (reportQueues == NULL) {
            fprintf(stderr, "create thread_queue error\n");
            exit(1);
        }

        pthread_t tid;
        int ret = pthread_create(&tid, NULL, store_main, reportQueues[i]);
        if (ret == -1)  {
            perror("pthread_create");
            exit(1);
        }

        pthread_detach(tid);
    }
}

2.Lars-reporterV0.1 reporter项目目录构建

int main(int argc, char **argv)
{
    event_loop loop;

    //加载配置文件
    config_file::setPath("./conf/lars_reporter.conf");
    std::string ip = config_file::instance()->GetString("reactor", "ip", "0.0.0.0");
    short port = config_file::instance()->GetNumber("reactor", "port", 7779);

    //创建tcp server
    tcp_server server(&loop, ip.c_str(), port);

    //添加数据上报请求处理的消息分发处理业务
    server.add_msg_router(lars::ID_ReportStatusRequest, get_report_status);

    //为了防止在业务中出现io阻塞，那么需要启动一个线程池对IO进行操作的，接受业务的请求存储消息
    create_reportdb_threads();

    //启动事件监听
    loop.event_process();

    return 0;
}
```

        这里主线程启动了线程池，根据配置文件的`db_thread_cnt`数量来开辟。每个线程都会执行`store_main`方法，我们来看一下实现

> lars_reporter/src/store_thread.cpp

```c
#include "lars.pb.h"
#include "lars_reactor.h"
#include "store_report.h"

struct Args
{
    thread_queue<lars::ReportStatusRequest>* first;
    StoreReport *second;
};

//typedef void io_callback(event_loop *loop, int fd, void *args);
void thread_report(event_loop *loop, int fd, void *args)
{
    //1. 从queue里面取出需要report的数据(需要thread_queue)
    thread_queue<lars::ReportStatusRequest>* queue = ((Args*)args)->first;
    StoreReport *sr = ((Args*)args)->second;

    std::queue<lars::ReportStatusRequest> report_msgs;

    //1.1 从消息队列中取出全部的消息元素集合
    queue->recv(report_msgs);
    while ( !report_msgs.empty() ) {
        lars::ReportStatusRequest msg = report_msgs.front();
        report_msgs.pop();

        //2. 将数据存储到DB中(需要StoreReport)
        sr->store(msg);
    }
}

3.Lars-ReporterV0.1 数据表和proto协议环境搭建

void *store_main(void *args)
{
    //得到对应的thread_queue
    thread_queue<lars::ReportStatusRequest> *queue = (thread_queue<lars::ReportStatusRequest>*)args;

    //定义事件触发机制
    event_loop loop;

    //定义一个存储对象
    StoreReport sr;

    Args callback_args;
    callback_args.first = queue;
    callback_args.second = &sr;

    queue->set_loop(&loop);
    queue->set_callback(thread_report, &callback_args);

    //启动事件监听
    loop.event_process();

    return NULL;
}
```

        每个线程都会绑定一个`thread_queue<lars::ReportStatusRequest>`,然后一个线程里面有一个loop,来监控消息队列是否有消息事件过来，如果有消息实现过来，针对每个消息会触发`thread_report()`方法，在`thread_report()`中，我们就直接将`lars::ReportStatusRequest`消息存储到db中。

        那么，由谁来给每个线程的`thread_queue`发送消息呢，就是agent/客户端发送的请求，我们在处理`lars::ID_ReportStatusRequest` 消息分发业务的时候调用`get_report_status()`来触发。

> lars_reporter/src/reporter_service.cpp

4.Lars-ReporterV0.1上报请求业务处理

```c
void get_report_status(const char *data, uint32_t len, int msgid, net_connection *conn, void *user_data)
{
    lars::ReportStatusRequest req;

    req.ParseFromArray(data, len);

    //将上报数据存储到db
    StoreReport sr;
    sr.store(req);

    //轮询将消息平均发送到每个线程的消息队列中
    static int index = 0;
    //将消息发送给某个线程消息队列
    reportQueues[index]->send(req);
    index ++;
    index = index % thread_cnt;
}
```

        这里的分发机制，是采用最轮询的方式，是每个线程依次分配，去调用`thread_queue`的`send()`方法，将消息发送给消息队列。

        最后我们进行测试，效果跟之前的效果是一样的。我们现在已经集成进来了存储线程池，现在就不用担心在处理业务的时候，因为DB等的io阻塞，使cpu得不到充分利用了。

###

5.Lars-ReporterV0.1上报请求模块的测试

# 六、Lars-Load Balance Agent负载代理

## 1) 简介

        一个服务称为一个模块，一个模块由modid+cmdid来标识
modid+cmdid的组合表示一个远程服务，这个远程服务一般部署在多个节点上

LB Agent以UDP方式为业务方提供：1、节点获取服务；2、节点调用结果上报服务

### 1.1 业务1-节点获取服务：

        业务方每次要向远程服务发送消息时，先利用modid+cmdid去向LB Agent获取一个可用节点，然后向该节点发送消息，完成一次远程调用；具体获取modid+cmdid下的哪个节点是由LB Agent负责的

### 1.2 业务2-节点调用结果上报服务

        对LB Agent节点的一次远程调用后，调用结果会汇报给LB Agent，以便LB Agent根据自身的LB算法来感知远程服务节点的状态是空闲还是过载，进而控制节点获取时的节点调度.

![4-Lars-agent](./pictures/4-Lars-agent.png)

6.Lars-ReporterV0.2开辟存储线程池-网络存储分离

LB Agent拥有5个线程，一个LB算法：

- UDP Server服务，并运行LB算法，对业务提供节点获取和节点调用结果上报服务；为了增大系统吞吐量，使用3个UDP Server服务互相独立运行LB算法：`modid+cmdid % 3 = i`的那些模块的服务与调度，由第`i+1`个UDP Server线程负责
- Dns Service Client：是dnsserver的客户端线程，负责根据需要，向dnsserver获取一个模块的节点集合（或称为获取路由）；UDP Server会按需向此线程的MQ写入获取路由请求，DSS Client将MQ到来的请求转发到dnsserver，之后将dnsserver返回的路由信息更新到对应的UDP Server线程维护的路由信息中
- Report Service Client：是reporter的客户端线程，负责将每个模块下所有节点在一段时间内的调用结果、过载情况上报到reporter Service端，便于观察情况、做报警；本身消费MQ数据，UDP Server会按需向MQ写入上报状态请求