气象数据中心项目，项目主语言为C++，涵盖多种后端技术栈.zip

# 气象数据中心实战项目 [TOC] ## 项目介绍 项目依托气象行业数据（数据总量大，且每天新增数据量大），模拟气象数据中心 该项目从数据源采集数据后存储到多个数据库中；业务系统通过数据库集群的总线访问数据 数据源：卫星接收站、雷达系统、国家基本站、区域气象站 业务系统：预警发布系统、预报制作系统、公共服务平台、数据共享平台 数据采集->数据处理->数据入库（数据库集群）->数据服务总线 主要功能子系统为： **1.数据采集子系统** 功能：把数据从数据源取出 三个模块：（1）ftp客户端，直接下载数据文件；（2）http客户端，采用http协议，从WEB服务接口数据；（3）直连数据库的数据源，从表中抽取数据 **2.数据处理和加工子系统** 功能：把各种格式的原始数据解码为xml格式的数据文件；对原始数据进行二次加工生成高可用的数据集。 **3.数据入库子系统** 功能：把数百种数据存储到数据中心的表中 **4.数据同步子系统** MySQL的高可用方案只能解决部分问题，且不够灵活，效率不高（只能满足数据库之间的复制不能满足表之间的复制）。本项目把核心数据库(Slave)表中的数据按条件和数据增量两种方式同步到业务数据库中。 项目数据库主要分为两种Master和Slave。 业务数据库主要由业务的需求决定（历史库（量大效率低）， 实时库（量小效率高）， 预报库， 台风库（量小效率高）， 接口库） **5.数据管理子系统** 功能：清理历史数据；把历史数据备份和归档 **6.数据交换子系统** 功能：把数据中心的数据从表中导出来生成数据文件。采用ftp协议，把数据文件推送到对方的ftp服务器。基于tcp协议的快速传输文件子系统 **7.数据服务总线子系统** 功能：用C++开发WEB服务，为业务提供数据访问接口，使用数据库连接池、线程池效率极高 **8.网络代理服务子系统** 功能：用于运维，利用I/O复用技术(select/poll/epoll) **项目重点和难点：**（1）服务程序的稳定性；（2）数据处理和服务的效率；（3）功能模块的通用性 ## 开发环境 CentOS7.x MySQL5.7 字符集utf-8 ## 项目框架  ## 服务程序常驻后台 保持服务程序稳定性，用守护进程监控服务程序的运行状态，如果服务程序故障，调度进程会重启服务程序 ### 服务程序的调度 利用linux信号和多进程 先执行fork函数，创建一个子进程，让子进程调用execl执行新的程序 新程序将替换子进程，不会影响父进程 在父进程中，可以调用wait函数等待新程序运行的结果，这样就可以实现调度的功能 ### 守护进程的实现 利用linux共享内存和linux信号量 // 查看共享内存： ipcs -m 删除共享内存： ipcrm -m shmid // 查看信号量： ipcs -s 删除信号量： ipcrm sem semid **信号量CSEM m_sem;** m_semid=semget(key,1,0666) 如果errno==2表示信号量不存在， 创建m_semid=**semget**(key,1,0666|IPC_CREAT|IPC_EXCL) // 信号量创建成功后，还需要把它初始化成value。 union semun sem_union; sem_union.val = value; // 设置信号量的初始**semctl**(m_semid,0,SETVAL,sem_union) // **加锁** CSEM::P 调用： **semop**(m_semid,&sem_b,1) 其中：struct **sembuf** sem_b; sem_b.sem_num = 0; sem_b.sem_op = -1; sem_b.sem_flg = m_sem_flg; **共享内存** **创建/获取**共享内存，键值为SHMKEYP，大小为MAXNUMP个st_procinfo结构体的大小。 **shmget**：m_shmid = shmget((key_t)SHMKEYP, MAXNUMP*sizeof(struct st_procinfo), 0666|IPC_CREAT) 将共享内存**链接**到当前进程的地址空间。 **shmat**： m_shm=(struct st_procinfo *)shmat(m_shmid, 0, 0); 把共享内存从当前进程中**分离**。 **shmdt** ：shmdt(m_shm) ## 数据采集子系统&数据交换子系统 ### ftp客户端下载模块 模块功能：已知服务器的文件路径名和保存到本地的文件路径名，**把服务器上某个目录的文件全部下载到本地目录**（可以指定文件名匹配规则）。文件下载成功后，ptype==1，服务器啥都不做，ptype==2，服务器删除文件，ptype==3，服务器备份文件 模块位置：/project/tools1/c/ftpgetfiles.cpp 模块实现： 1. 基于Cftp类，头文件在/project/public/_ftp.h中 结构体st_fileinfo中包含文件名filename，文件时间mtime 容器vlistfile2存放服务器待下载文件信息；容器vlistfile1存放已经传输完成的文件信息； 容器vlistfile3存放本地以及下载的但服务器中还有的文件；容器vlistfile4存放服务器中新增文件 listfilename中存放服务器文件；okfilename中存放已经传输完成的文件；strremotefilenamebak备份目录 2. 输入参数写成xml格式方便应对众多输入参数。利用**GetXMLBuffer**函数读取输入参数 2. 创建心跳信息：**CPActive**类中的**AddPInfo**(starg.timeout, starg.pname)函数 3. 登录ftp服务器：**ftp.login**(starg.host, starg.username, starg.password, starg.mode) 4. 进入ftp服务器存放文件的目录**ftp.chdir**(starg.remotepath)，（调用FtpChdir） 5. 调用**ftp.nlist()**方法列出服务器目录中的文件，结果存放到本地文件listfilename中： ftp.nlist(".", starg.listfilename)，其中有MKDIR()和FtpNlst(listfilename,remotedir,m_ftpconn) 6. LoadListFile加载文件函数，将上述ftp.nlist()方法获取到的list文件加载到容器vlistfile2中（1）打开本地文件File.Open（listfilename）（2）读取本地文件中的一行到File.Fgets（3）MatchStr匹配满足条件的文件（4）当ptype为1，checktime为true时，调用ftp.mtime（底层调用FtpModDate函数）获取ftp服务器文件时间放入ftp.m_mtime中 7. 如果ptype==1（1）加载okfilename文件中的内容到容器vlistfile1中（2）比较vlistfile2和vlistfile1，得到vlistfile3和vlistfile4（3）把vlistfile4中的内容复制到vlistfile2中：vlistfile2.clear(); vlistfile2.swap(vlistfile4); 8. 遍历容器vlistfile2，调用**ftp.get()**方法从服务器下载文件：ftp.get(strremotefilename, strlocalfilename) 9. 如果ptype==1，把下载成功的文件记录追加到okfilename文件中 10. 如果ptype==2，使用ftp.ftpdelete()删除服务器文件 11. 如果ptype==3，使用ftp.ftprename()服务器文件转存到服务器备份目录 ### ftp客户端下载模块 ## 数据处理和统计系统 数据处理： 任务：把各种格式的数据文件转换为xml格式的文件，再交给入库程序 步骤：1. 读取目录中的文件 2.解析文件的内容 3.生成xml文件格式的文件 数据统计： 任务：把采集到的数据进行二次加工，生成新的数据产品，最大化数据价值 ## 数据入库子系统 ## 数据同步子系统 MySQL高可用方案的不足： 1.Master（写）和Slave（读）的表结构和数量必须一致 2.非主从关系的数据库不能进行复制 3.不够灵活，效率也不高 我们的做法： 基于MySQL的Federated存储引擎（Oracle dblink）把远程数据库同步到本地（实现数据库中的互相访问） ### 刷新同步模块 针对需要覆盖本地表的情况，开发了**刷新同步模块**，模块位置：/project/tools1/c/syncupdate.cpp 模块实现： 若需要同步的数据量少，采用一次性全部同步：1. delete from (localtname) (where) 删除本地表中的