基于IPV4的流媒体广播

这个项目和webserver挺像的，但是这个是基于UDP的，实现了一个组播。有服务端和客户端两个。

服务端：主要是不停的向外广播自己的节目单和节目内容（没错，就算没有客户端开启，也会一直广播）

客户端：主要是根据自己的需求订阅服务端的节目单，然后接收相应的节目数据，传递给ffmpeg，让ffmpeg来解析一下内容，然后播放出来
（是不是感觉挺高大上的，其实不是，只是简单的接收所有的数据，然后判断这个数据里面的ID是不是自己选择的就好，是自己选择的就保留，不是就丢弃）

运行起来

原项目git：https://github.com/litbubo/Streaming_media_broadcasting_system_based_on_IPv4

环境: Ubuntu 20.04 ffmpeg

还需要下载一个ffmepg包，这个是啥包我忘记了，但是在运行client的时候会发现只会接收一个UDP包的数据

这是因为缺少了这个包。接收数据之后会馈入ffmpeg里面，因为缺少了这个包，所以才会只接收一个数据之后就断掉

终端上会有提示，到时候直接ubuntu命令下载即可

// 终端-1 server
cd server
make
./server

// 终端-2 client
cd client
./start.sh

(这里不是简单的接收数据，ffmpeg是可以正常播放出来音乐的，放不出来可能是虚拟机的问题，我这边是可以正常放出来的)

结构

6_IPV4UDP/
├── client
│   ├── client.c
│   ├── client.h
│   ├── Makefile
│   └── start.sh
├── include
│   └── protocol.h
├── medialib
│   ├── ch1
│   │   ├── desc.txt
│   │   ├── 世界这么大还是遇见你.mp3
│   │   └── 起风了.mp3
│   ├── ch2
│   │   ├── desc.txt
│   │   ├── 带你去旅行.mp3
│   │   └── 再见只是陌生人.mp3
│   ├── ch3
│   │   ├── desc.txt
│   │   ├── Summertime Sadness.mp3
│   │   └── 学猫叫.mp3
│   └── ch4
├── server
│   ├── channel.c
│   ├── channel.h       //每个频道的广播
│   ├── list.c
│   ├── list.h          //节目单广播
│   ├── Makefile
│   ├── medialib.c
│   ├── medialib.h      //medialib转化为广播频道
│   ├── server.c
│   ├── server_conf.h   //服务端的一些基础的配置
│   ├── threadpool.c
│   ├── threadpool.h    //线程池
│   ├── tokenbucket.c
│   └── tokenbucket.h   //令牌桶，负责流量控制
├── README.md
├── LICENSE
└── Streaming_media_broadcasting_system_based_on_IPv4.code-workspace

复现

协议

protocol.h 文件，这里面定义了很多的基本的配置，比如：多播组的地址，默认的端口号，还有节目单，频道的格式等。

代码:这里面不涉及啥复杂的操作，看一看即可

#ifndef __PROTOCOL_H__
#define __PROTOCOL_H__

#include<stdint.h>
#include<stdio.h>

typedef uint8_t chnid_t;    //定义频道ID类型为8位无符号整数，范围0~255，节省空间

#define DEFAULT_MGROUP     "224.2.2.2" // 多播组
#define DEFAULT_RECVPORT   "1989"      // 默认端口
#define CHANNR             100         // 最大频道数量

#define LISTCHNID          0           // 约定 0 号为节目单频道
#define MINCHNID           1           // 最小广播频道号
#define MAXCHNID           (CHANNR - MINCHNID)  // 最大广播频道号 100 - 1

#define MSG_CHANNEL_MAX    (65536U - 20U - 8U)  // 最大频道数据包  20U IP头   8U UDP头
#define MAX_CHANNEL_DATA   (MSG_CHANNEL_MAX - sizeof(chnid_t)) //MSG_CHANNEL_MAX去除掉一个 

#define MSG_LISTCHN_MAX     (65536U - 20U - 8U) // 最大节目单数据包  20U IP头   8U UDP头
#define MAX_LISTCHN_DATA    (MSG_LISTCHN_MAX - sizeof(chnid_t))


/* 频道包，第一字节描述频道号，data[0]在结构体最后作用为变长数组，根据malloc到的实际内存大小决定 */
typedef struct msg_channel_t{
    chnid_t chnid;             // must between MINCHNID MAXCHNID 频道号（1字节） 取值范围MINCHNID~MAXCHNID
    uint8_t data[0];           // 柔性数组，实际数据存储位置（长度由MAX_DATA决定）
} __attribute__((packed)) msg_channel_t;  //不做对其，因为chnid就只有一个字节， data 通常会很大（这个我还不是很理解，对其的话chnid不是最多也就是4个字节码，感觉有没有这个没多大差）

/* 单个节目信息，第一字节描述频道号，第二三字节为本条信息的总字节数，desc[0]为变长数组 */
typedef struct msg_listdesc_t{
    chnid_t chnid;             // 频道号（1字节）
    uint16_t deslength;        // 自述包长度 描述信息长度
    uint8_t desc[0];           // 柔性数组，存储频道描述文本（UTF-8格式）
} __attribute__((packed)) msg_listdesc_t;


/* 节目单数据包，第一字节描述频道号，list[0]为变长数组，存储msg_listdesc_t变长内容 */
typedef struct msg_list_t{
    chnid_t chnid;           // 频道号（1字节）
    msg_listdesc_t list[0];  // 柔性数组，包含多个节目条目
}__attribute__((packed)) msg_list_t;

#endif

服务端：

需要完成的任务:

1.将medialib目录下的文件转化为广播频道

2.广播节目单

3.广播频道内容

整体的复现思路：

1.server_conf.h 里面有一些服务端的基础配置

2.threadpool 这个比较独立，就是基础的线程池，用来管理线程。不依赖于其他模块

3.tokenbucket 令牌桶，用来控制流量，防止流量过大

4.medialib 主要是将medialib目录下的文件转化为广播频道，然后通过list广播节目单，通过channel广播频道内容

5.channel 主要是广播频道内容

6.list 主要是广播节目单

7.server 主要是管理线程池，然后调用list和channel来广播节目单和频道内容

按照这个思路，依次完成各个模块，然后最后整合起来，就可以完成服务端了

server_conf.h

没啥好说的，直接看代码

#ifndef __SERVER_CONF_H__
#define __SERVER_CONF_H__

#define DEFAULT_MEDIADIR "/var/medialib"    // 默认本地媒体库路径
#define DEFAULT_IF "ens33"                  // 默认网卡

#include "threadpool.h"

//运行模式，守护进程会把进程放到后台
enum RNUMODE                                // 运行模式    
{
    RUN_DAEMON = 0,                         // 守护进程
    RUN_FOREGROUND                          // 前台运行
};

typedef struct server_conf_t                // 配置文件,
{
    char *mgroup;                           // [字符串] 组播组IP地址（如："239.0.0.1"）
    char *rcvport;                          // [字符串] 接收端口号（如："8080"）
    char *media_dir;                        // [字符串] 媒体文件存储目录路径
    enum RNUMODE runmode;                   // [字符] 运行模式（使用上述枚举值：RUN_DAEMON/RUN_FOREGROUND）
    char *ifname;                           // [字符串] 网络接口名称（用于绑定组播）
} server_conf_t;

extern server_conf_t server_conf;           // 配置文件
extern int serversd;                        // 服务端套接字
extern struct sockaddr_in sndaddr;          // 发送目的地址
extern ThreadPool_t *pool;                  // 线程池对象

#endif // !__SERVER_CONF_H__

threadpool

这个要说一下，我之前做过webserver的项目，那个里面的线程池比较简单，就是单纯的addtask，然后执行任务就可以了，但是这个项目里面的线程池构造的很巧妙，如果看了代码就可以很好理解。

首先最基本的功能肯定是有的，就是addTask，然后执行任务

这里面还附加了另外的功能，就是设置了一个管理的线程，这个管理的线程用于管理线程池中线程的数量，当任务不多的时候，没必要保存那么多线程，这个时候就可以kill掉一些，当然会有一个最小的数量。当任务很多的时候，就需要额外申请一些线程，但是这个数量也是有限制的。

这样来实现线程数量的动态管理，不会因为空线程而浪费太多的CPU资源

有几个注意的点:

1.NUMSTEP, 每次进行添加和删除线程都会增加或者删除 NUMSTEP 多的线程。不会说一下子减少到最少，或者是一下子添加到最大。拿减少的来举例：如果检测到很多线程空闲，我先减少NUMSTEP个线程，然后再次检测，如果还是空闲，我再减少NUMSTEP多线程。

2.numExit 这个设置的就很巧妙，就是如果我要删除掉一个线程，我就把numExit设置为1，并且notify一下，线程在working (在working函数里面) 的时候，先检测一下numExit是不是1 (这里肯定是要加锁的哈)，如果是1，那么我就不去task，直接退出（自杀）就好了，顺便把numExit–（上锁哈）。是不是感觉很巧妙

3.这里面有个独立出来的manage的线程哈，这个别忘记了，执行的是manager函数，里面就涉及到了之前说的一些线程的增加与删除等操作

代码:

//threadpool.h
#ifndef __THREADPOOL_H__
#define __THREADPOOL_H__

// #define DEBUG                                // 定义宏，DEBUG模式，打印尽可能多的的log信息，注释则不打印

typedef void ThreadPool_t;                      // 对外隐藏ThreadPool_t内部实现

/*
 * @name            : threadpool_create
 * @description		: 线程池创建函数，创建一个线程池
 * @param - min 	: 最小线程池数
 * @param - max 	: 最大线程池数
 * @param - queueCapacity : 最大任务队列数
 * @return 			: 失败返回 NULL，成功返回线程池对象地址
 */
ThreadPool_t* threadpool_create(int min,int max,int queueCapacity);                     // 新创建一个线程池

/*
 * @name            : threadpool_addtask
 * @description		: 任务队列添加任务函数，添加一个任务
 * @param - argPool : 传入需要添加任务的线程池
 * @param - function: 任务函数 void (*)(void *, volatile int *) 函数指针  【返回类型 (*指针变量名)(参数列表)】
 * @param - arg     : 任务函数参数
 * @return 			: 失败返回 -1，成功返回 0
 */
int threadpool_addtask(ThreadPool_t *, void (*)(void *, volatile int *), void *);       // 向任务队列添加一个任务


/*
 * @name            : threadpool_destroy
 * @description		: 线程池销毁函数，销毁一个线程池
 * @param - argPool : 传入需要销毁的线程池对象
 * @return 			: 失败返回 -1，成功返回 0
 */
int threadpool_destroy(ThreadPool_t *);                                                 // 销毁一个线程池

/*
 * @name            : threadexit_unlock
 * @description		: 线程退出函数，并将该线程 ID 从工作者线程数组中删除
 * @param - argPool : 传入线程池对象
 * @return 			: 无
 */
void threadexit_unlock(ThreadPool_t *);                                                 // 线程退出函数

/*
 * @name            : get_thread_live
 * @description		: 获取线程池中存活线程数
 * @param - argPool : 传入线程池对象
 * @return 			: 线程池中存活线程数
 */
int get_thread_live(ThreadPool_t *);                                                   // 获得线程池中的存活线程数

/*
 * @name            : get_thread_busy
 * @description		: 获取线程池中忙线程数
 * @param - argPool : 传入线程池对象
 * @return 			: 线程池中忙线程数
 */
int get_thread_busy(ThreadPool_t *);                                                   // 获得线程池中的忙线程数
#endif

// threadpool.cs
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <sys/select.h>
#include <sys/types.h>
#include <unistd.h>
#include <errno.h>
#include <signal.h>
#include <syslog.h>

#include "threadpool.h"

#define NUMSTEP 5      // 线程增减步长
#define BUFSIZE 1024   // 状态显示缓冲区大小

typedef struct Task_t {
    void (*function)(void *, volatile int *);   // 任务函数指针
    void *arg;                                  // 任务参数
} Task_t;

// 定义线程池类型
struct ThreadPool_t{ 
    // 任务队列相关
    Task_t *taskQueue;          // 任务队列数组
    int queueCapacity;          // 任务队列最大容量
    int queueSize;              // 任务队列当前任务数
    int queueRear;              // 队尾
    int queueFront;             // 队头

    // 线程管理相关
    pthread_t managerID;        // 管理者线程ID
    pthread_t *workerIDs;       // 工作者线程ID数组
    int numMax;                 // 工作者线程最大的线程数
    int numMin;                 // 工作者线程最小的线程数
    int numLive;                // 工作者线程存活的线程数
    int numBusy;                // 工作者线程忙的线程数
    int numExit;                // 工作者线程需要退出的线程数

    // 同步机制
    pthread_mutex_t mutexPool;  // 线程池锁
    pthread_mutex_t mutexBusy;  // 忙线程数锁
    pthread_cond_t notFull;     // 非满条件变量，用于唤醒生产者(添加任务函数)
    pthread_cond_t notEmpty;    // 非空条件变量，用于唤醒消费者(工作者线程)

    volatile int shutstatus;             // 线程池状态，0 打开，-1 关闭
};

// DEBUG 模式下 调试函数：图形化显示线程状态
#ifdef DEBUG
static void printstatus(ThreadPool_t *argPool){
    struct ThreadPool_t *pool = (struct ThreadPool_t *)argPool;
    int numLive, numBusy;
    int i;
    char buf[BUFSIZE];
    memset(buf, 0, BUFSIZE);

    // 获取线程状态
    numLive = get_thread_live(pool);
    numBusy = get_thread_busy(pool);
    // 构建状态字符串：'+'表示忙碌，'-'表示空闲
    for (i = 0; i < numBusy; i++)
        strcat(buf, "+");
    for (i = 0; i < numLive - numBusy; i++)
        strcat(buf, "-");
    // 输出格式：[++++++++++-----] : busy=10, live=15
    fprintf(stdout, "[ %s ] : busy == %d, live == %d\n", buf, numBusy, numLive);
}
#endif // DEBUG

/*
 * @name            : working
 * @description		: 工作者线程任务函数，负责从任务队列中取出任务并执行
 * @param - arg 	: 传入线程池对象
 * @return 			: NULL
 */
static void *working(void *arg){
    struct ThreadPool_t *pool = (struct ThreadPool_t *)arg;
    Task_t task;
    sigset_t set;
    // 阻塞所有信号，避免工作线程处理信号
    sigfillset(&set);
    sigprocmask(SIG_BLOCK, &set, NULL);

    while (1){
        pthread_mutex_lock(&pool->mutexPool);  // 加线程池锁
        while (pool->queueSize == 0 && pool->shutstatus == 0){
            pthread_cond_wait(&pool->notEmpty, &pool->mutexPool); // 阻塞直到任务队列不为空
            // 被唤醒后检查是否需要退出线程
            if (pool->numExit > 0) {
                pool->numExit--;
                if (pool->numLive > pool->numMin){
                    pool->numLive--;
                    pthread_mutex_unlock(&pool->mutexPool);
                    threadexit_unlock(pool); // 执行线程退出流程
                }
            }
        }

        if (pool->shutstatus == -1) // 若线程池已经关闭，线程自杀
        {
            pthread_mutex_unlock(&pool->mutexPool);
            threadexit_unlock(pool);
        }

        // 从队列之中取任务
        task.function = pool->taskQueue[pool->queueFront].function;      // 取出任务
        task.arg = pool->taskQueue[pool->queueFront].arg;
        memset(&pool->taskQueue[pool->queueFront], 0, sizeof(Task_t));   // 从队列取出任务后，将队列中相应任务清空
        pool->queueFront = (pool->queueFront + 1) % pool->queueCapacity; // 移动队头指针
        pool->queueSize--;                                               // 任务队列中任务数量-1
        pthread_cond_signal(&pool->notFull);                             // 唤醒任务生产者 //取出来一个任务后有空位置了自然要处理一下
        pthread_mutex_unlock(&pool->mutexPool);

        pthread_mutex_lock(&pool->mutexBusy); // 加锁，改变线程池中忙线程数
        pool->numBusy++;
        pthread_mutex_unlock(&pool->mutexBusy);
#ifdef DEBUG
        fprintf(stdout, "[thread = %ld] is going to work...\n", pthread_self());
#endif // DEBUG

        task.function(task.arg, &(pool->shutstatus)); // 执行任务

#ifdef DEBUG
        fprintf(stdout, "[thread = %ld] is done work...\n", pthread_self());
#endif // DEBUG
        
        free(task.arg); // 释放任务资源

        syslog(LOG_INFO, "thread [%ld] is free successful...", pthread_self());

#ifdef DEBUG
        fprintf(stdout, "[thread = %ld] is free successful...\n", pthread_self());
#endif // DEBUG

        task.function = NULL;
        task.arg = NULL;

        pthread_mutex_lock(&pool->mutexBusy);
        pool->numBusy--;
        pthread_mutex_unlock(&pool->mutexBusy);
        sched_yield(); // 出让调度器给其他线程
    }
    pthread_exit(NULL);
}


/*
 * @name            : manager
 * @description		: 管理者线程任务函数，负责监视、增加和减少线程池中线程的存活线程数量
 * @param - arg 	: 传入线程池对象
 * @return 			: NULL
 */
static void *manager(void *arg){
    struct ThreadPool_t *pool = (struct ThreadPool_t *)arg;
    struct timeval tv;
    int numLive, numBusy, queueSize;
    int i, count;
    sigset_t set;

    // 阻塞所有信号
    sigfillset(&set);
    sigprocmask(SIG_BLOCK, &set, NULL);

    while (pool->shutstatus == 0) {  // 线程池运行中循环
        // 定时2s，可根据实际场景改变
        tv.tv_sec = 2;
        tv.tv_usec = 500000;
        select(0, NULL, NULL, NULL, &tv);  // select作为延时函数，替换sleep，保证线程安全

        // 获取当前状态（减少锁持有时间）
        pthread_mutex_lock(&pool->mutexPool);
        numLive = pool->numLive;
        queueSize = pool->queueSize;
        pthread_mutex_unlock(&pool->mutexPool);

        pthread_mutex_lock(&pool->mutexBusy);
        numBusy = pool->numBusy;
        pthread_mutex_unlock(&pool->mutexBusy);

        // 动态扩容逻辑 
        count = 0;  
        if ((numLive < queueSize || numBusy > numLive*0.8) && numLive < pool->numMax) { // 当存活线程数小于待取任务数量，并且小于最大线程数
            pthread_mutex_lock(&pool->mutexPool); // 添加 NUMSTEP 步长的线程
            // 以NUMSTEP为步长创建新线程
            for (int i = 0; i < pool->numMax && count < NUMSTEP && pool->numLive < pool->numMax; i++) {
                if (pool->workerIDs[i] == 0) {  // 找到空闲位置
                    pthread_create(&pool->workerIDs[i], NULL, working, pool);
                    count++;
                    pool->numLive++;
                }
            }
            pthread_mutex_unlock(&pool->mutexPool);

        }
        //释放多余线程
        if (numBusy * 2 < numLive && numLive > pool->numMin) { // 当忙线程数 * 2小于存活线程数，并且存活的线程大于最小线程数
            pthread_mutex_lock(&pool->mutexPool);
            pool->numExit = NUMSTEP;  // 设置退出数量
            pthread_mutex_unlock(&pool->mutexPool);
            for (i = 0; i < NUMSTEP; i++){
                pthread_cond_signal(&pool->notEmpty); // 唤醒工作线程，让其自杀
            }
        }

#ifdef DEBUG
        printstatus(pool); // 打印线程池中线程信息
#endif // DEBUG
        sched_yield();     // 出让调度器  
    }
    pthread_exit(NULL);
}

/*
 * @name            : threadpool_create
 * @description		: 线程池创建函数，创建一个线程池
 * @param - min 	: 最小线程池数
 * @param - max 	: 最大线程池数
 * @param - queueCapacity : 最大任务队列数
 * @return 			: 失败返回 NULL，成功返回线程池对象地址
 */
ThreadPool_t *threadpool_create(int min, int max, int queueCapacity){
    struct ThreadPool_t* pool = malloc(sizeof(struct ThreadPool_t));
    int i;
    do{
        if(pool == NULL){
            syslog(LOG_ERR, "threadpool malloc() : %s", strerror(errno));
            break; // 申请内存失败就跳过剩下的初始化
        }
        // 初始化任务队列
        pool->taskQueue = malloc(sizeof(Task_t) * queueCapacity);
        if (pool->taskQueue == NULL){
            syslog(LOG_ERR, "taskQueue malloc() : %s", strerror(errno));
            break;
        }
        memset(pool->taskQueue, 0, sizeof(Task_t) * queueCapacity); //清空一下
        pool->queueCapacity = queueCapacity; // 各个成员的初始化
        pool->queueSize = 0;
        pool->queueRear = 0;
        pool->queueFront = 0;

        // 初始化工作线程数组
        pool->workerIDs = malloc(sizeof(pthread_t) * max);
        if (pool->workerIDs == NULL)
        {
            syslog(LOG_ERR, "workerIDs malloc() : %s", strerror(errno));
            break;
        }
        memset(pool->workerIDs, 0, sizeof(pthread_t) * max); //清空一下
        pool->numMax = max;
        pool->numMin = min;
        pool->numLive = min;
        pool->numBusy = 0;
        pool->numExit = 0;

        if (pthread_mutex_init(&pool->mutexPool, NULL) != 0 || // 初始化锁和条件变量
            pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||
            pthread_cond_init(&pool->notFull, NULL) != 0 ||
            pthread_cond_init(&pool->notEmpty, NULL) != 0)
        {
            syslog(LOG_ERR, "lock init failed ...");
            break;
        }

        pool->shutstatus = 0;  // 运行状态 // 开启线程池

        pthread_create(&pool->managerID, NULL, manager, pool); // 创建管理者线程

        for(i=0;i<min;i++){
            if (pool->workerIDs[i] == 0){
                pthread_create(&pool->workerIDs[i], NULL, working, pool); // 创建工作者线程
            }
        }

        return pool;

    } while (0);

    if (pool != NULL && pool->workerIDs != NULL) // 申请内存失败跳转到这里开始，依次析构 //这边是错误处理
    free(pool->workerIDs);
    if (pool != NULL && pool->taskQueue != NULL)
        free(pool->taskQueue);
    if (pool != NULL)
        free(pool);
    return NULL;
    
}


/*
 * @name            : threadpool_destroy
 * @description		: 线程池销毁函数，销毁一个线程池
 * @param - argPool : 传入需要销毁的线程池对象
 * @return 			: 失败返回 -1，成功返回 0
 */
int threadpool_destroy(ThreadPool_t *argPool){
    int i;
    struct ThreadPool_t *pool = (struct ThreadPool_t *)argPool;

    if (pool == NULL)
    {
        syslog(LOG_ERR, "thread pool is not existed ...");
        return -1;
    }

    pool->shutstatus = -1;  // 触发关闭

    sleep(2);
    pthread_join(pool->managerID, NULL); // join掉managerID线程

    for (i = 0; i < pool->numLive; i++)
    {
        pthread_cond_signal(&pool->notEmpty); // 唤醒所有存活线程，让其自杀
    }
    //信号量销毁
    pthread_mutex_destroy(&pool->mutexBusy);
    pthread_mutex_destroy(&pool->mutexPool);
    pthread_cond_destroy(&pool->notEmpty);
    pthread_cond_destroy(&pool->notFull);

    // free掉空间
    if (pool != NULL && pool->workerIDs != NULL)
        free(pool->workerIDs);
    if (pool != NULL && pool->taskQueue != NULL)
        free(pool->taskQueue);
    if (pool != NULL)
        free(pool);
    pool = NULL;

    syslog(LOG_INFO, "thread pool is going to be destroyed...");
#ifdef DEBUG
    fprintf(stdout, "thread pool is going to be destroyed...\n");
#endif // DEBUG
    return 0;
}

/*
 * @name            : threadpool_addtask
 * @description		: 任务队列添加任务函数，添加一个任务
 * @param - argPool : 传入需要添加任务的线程池
 * @param - function: 任务函数
 * @param - arg     : 任务函数参数
 * @return 			: 失败返回 -1，成功返回 0
 */
int threadpool_addtask(ThreadPool_t *argPool, void (*function)(void *, volatile int*), void *arg){
    struct ThreadPool_t *pool = (struct ThreadPool_t *)argPool;
    pthread_mutex_lock(&pool->mutexPool);
    while(pool->queueSize == pool->queueCapacity && pool->shutstatus == 0){
        pthread_cond_wait(&pool->notFull, &pool->mutexPool); // 阻塞直到等待任务队列不为满
    }
    if(pool->shutstatus == -1){
        syslog(LOG_INFO, "thread pool has been shutdown ...");
        pthread_mutex_unlock(&pool->mutexPool);
        return -1;
    }

    pool->taskQueue[pool->queueRear].function = function; // 将任务存储到任务队列中
    pool->taskQueue[pool->queueRear].arg = arg;
    pool->queueRear = (pool->queueRear + 1) % pool->queueCapacity; // 移动队尾指针
    pool->queueSize++;
    pthread_cond_signal(&pool->notEmpty); // 队列不为空，唤醒工作者线程
    pthread_mutex_unlock(&pool->mutexPool);
    return 0;
}


/*
 * @name            : threadexit_unlock
 * @description		: 线程退出函数，并将该线程 ID 从工作者线程数组中删除
 * @param - argPool : 传入线程池对象
 * @return 			: 无
 */
void threadexit_unlock(ThreadPool_t *argPool){
    int i;
    struct ThreadPool_t *pool = (struct ThreadPool_t *)argPool;
    pthread_t tmptid = pthread_self();
    for (i = 0; i < pool->numMax; i++)
    {
        if (pool->workerIDs[i] == tmptid)
        {
            pool->workerIDs[i] = 0;
            break;
        }
    }
    syslog(LOG_INFO, "thread [%ld] is going to exit...", tmptid);
#ifdef DEBUG
    fprintf(stdout, "[thread = %ld] is going to exit...\n", tmptid);
#endif // DEBUG
    pthread_exit(NULL);
}

/*
 * @name            : get_thread_live
 * @description		: 获取线程池中存活线程数
 * @param - argPool : 传入线程池对象
 * @return 			: 线程池中存活线程数
 */
int get_thread_live(ThreadPool_t *argPool)
{
    struct ThreadPool_t *pool = (struct ThreadPool_t *)argPool;
    int num;
    pthread_mutex_lock(&pool->mutexPool);
    num = pool->numLive;
    pthread_mutex_unlock(&pool->mutexPool);
    return num;
}

/*
 * @name            : get_thread_busy
 * @description		: 获取线程池中忙线程数
 * @param - argPool : 传入线程池对象
 * @return 			: 线程池中忙线程数
 */
int get_thread_busy(ThreadPool_t *argPool)
{
    struct ThreadPool_t *pool = (struct ThreadPool_t *)argPool;
    int num;
    pthread_mutex_lock(&pool->mutexBusy);
    num = pool->numBusy;
    pthread_mutex_unlock(&pool->mutexBusy);
    return num;
}

tokenbucket

令牌桶，涉及到一些流量控制，就是我们在读mp3媒体文件的时候，不是要传递一个读取多少字节的参数吗？ 这个时候咱们先从令牌桶里面取令牌，如果说令牌数量很充足，就是大于size (size是咱们想要读的字节) 的话就取出size个令牌，然后读size个字节。但是如果说令牌桶里面取令牌不足size，比如只有size1这么多了，那么咱们读数据的时候就取出size1这么多字节。

这里面有个计时器，每一秒会使得令牌桶里面的令牌的数量增加

struct tokenbt_t{
    int cps;                    // 步长
    int burst;                  // 上限
    int token;                  // 拥有的令牌
    int pos;                    // 自述数组下标 在全局job数组中的位置索引,不是只有一个桶
    pthread_mutex_t mutex;
    pthread_cond_t cond;
};

这里看一下单个令牌桶的结构

注意的点：

1.这个定时器是额外定义的线程，他不会占用主线程。

2.并不是只有一个令牌桶，这里设计了一个令牌桶的数组 job数组。里面存储了好多令牌桶，这样针对每个频道，我们都可以为其分配一个令牌桶。让每个频道独立起来。而不是所有的频道公用一个令牌桶。

3.注意这个令牌桶使用的位置哈，并不是sendto和recvfrom这两个函数用，而是在读取mp3文件的时候来用，就是通过控制文件读取的速度来控制sendto发送数据包的速度。

4.这里有两种类型的锁，一个是job队列的锁，就是对job队列进行操作的时候必须得加锁。还有一个是桶内部的锁，看上面的结构，是不是每个桶都有一个锁，这是为了在实现安全的修改每个桶里面的数据。

#ifndef __TOKENBUCKET_h__
#define __TOKENBUCKET_h__

#define TOKENBUCKET_MAX 1024    //最大令牌桐对象数量

typedef void tokenbt_t;         //对外隐藏内部实现细节

/*
 * @name            : tokenbt_init
 * @description		: 令牌桶初始化，初始化一个令牌桶对象
 * @param - cps     : 步长 每一s增加多少令牌数量
 * @param - burst   : 上限零牌数
 * @return 			: 成功返回令牌桶对象，失败返回 NULL
 */
tokenbt_t *tokenbt_init(int, int);

/*
 * @name            : tokenbt_fetchtoken
 * @description		: 从令牌桶对象中取令牌
 * @param - token   : 令牌桶对象
 * @param - size    : 要取的零牌数
 * @return 			: 成功返回令牌，失败返回 -EINVAL
 */
int tokenbt_fetchtoken(tokenbt_t *, int);  


/*
 * @name            : tokenbt_checktoken
 * @description		: 检查令牌桶对象的令牌数
 * @param - token   : 令牌桶对象
 * @return 			: 成功返回令牌，失败返回 -EINVAL
 */
int tokenbt_checktoken(tokenbt_t *);

/*
 * @name            : tokenbt_returntoken
 * @description		: 给令牌桶对象归还令牌
 * @param - token   : 令牌桶对象
 * @param - size    : 令牌
 * @return 			: 成功返回归还的令牌，失败返回 -EINVAL
 */
int tokenbt_returntoken(tokenbt_t *, int);

/*
 * @name            : tokenbt_destroy
 * @description		: 销毁单个令牌桶对象
 * @param - token   : 令牌桶对象
 * @return 			: 成功返回 0，失败返回 -EINVAL
 */
int tokenbt_destroy(tokenbt_t *); 

/*
 * @name            : tokenbt_destroy_all
 * @description		: 销毁所有令牌桶对象
 * @param           : 无
 * @return 			: 成功返回 0
 */
int tokenbt_destroy_all(); 

/*
 * @name            : tokenbt_shutdown
 * @description		: 关闭令牌流控功能模块
 * @param           : 无
 * @return 			: 成功返回 0
 */
int tokenbt_shutdown(); 



#endif // !__TOKENBUCKET_h__

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <signal.h>
#include <syslog.h>

#include "tokenbucket.h"

struct tokenbt_t{
    int cps;                    // 步长
    int burst;                  // 上限
    int token;                  // 拥有的令牌
    int pos;                    // 自述数组下标 在全局job数组中的位置索引,不是只有一个桶
    pthread_mutex_t mutex;
    pthread_cond_t cond;
};

//现在有TOKENBUCKET_MAX个桶
static struct tokenbt_t *token_pool[TOKENBUCKET_MAX];

// 保护job数组的互斥锁，防止多线程并发修改
static pthread_mutex_t mutex_pool = PTHREAD_MUTEX_INITIALIZER; 
// 确保init模块只会加载一次
static pthread_once_t once_init = PTHREAD_ONCE_INIT;
// 定时器的线程ID
static pthread_t tid;

/*
 * @name            : woking
 * @description		: 任务函数，负责固定时间给令牌桶添加令牌数
 * @param - arg     : NULL
 * @return 			: NULL
 */
static void *woking(void *arg){
    int i;
    struct timeval tv;
    sigset_t set;

    sigfillset(&set);
    sigprocmask(SIG_BLOCK, &set, NULL);     // 屏蔽所有信号，只留主线程处理信号即可

    while(1){
        pthread_mutex_lock(&mutex_pool); //对token_pool加锁
        for(i=0;i<TOKENBUCKET_MAX;i++){
            if(token_pool[i]!=NULL){
                pthread_mutex_lock(&token_pool[i]->mutex);
                token_pool[i]->token += token_pool[i]->cps;
                if (token_pool[i]->token > token_pool[i]->burst){ //已经到最大值了
                    token_pool[i]->token = token_pool[i]->burst;
                }
                pthread_cond_broadcast(&token_pool[i]->cond); //notifyAll
                pthread_mutex_unlock(&token_pool[i]->mutex);
            }
        }
        pthread_mutex_unlock(&mutex_pool);
        tv.tv_sec = 1;
        tv.tv_usec = 0;
        select(0, NULL, NULL, NULL, &tv);       // 安全定时 1 s
    }
    pthread_exit(NULL);
}

/*
 * @name            : get_free_pos_unlocked
 * @description		: 任务队列添加任务函数，添加一个任务
 * @param           : 无
 * @return 			: 失败返回 -1，成功返回 找到的空闲数组位置
 */
static int get_free_pos_unlocked(){  //找到token_pool之中第一个空闲值
    int i;
    for(i =0;i<TOKENBUCKET_MAX;i++){
        if(token_pool[i]==NULL){
            return i;
        }
    }
    return -1;
}

/*
 * @name            : module_unload
 * @description		: 令牌桶模块卸载
 * @param           : 无
 * @return 			: 无
 */
static void module_unload(){
    pthread_cancel(tid);        // 取消任务函数线程(负责固定时间给令牌桶添加令牌数)
    pthread_join(tid, NULL);
    tokenbt_destroy_all();
    syslog(LOG_INFO, "free job is done ...");
}

/*
 * @name            : module_load
 * @description		: 令牌桶模块加载
 * @param           : 无
 * @return 			: 无
 */
static void module_load(){
    pthread_create(&tid, NULL, woking, NULL);  //创建计时器线程
}

/*
 * @name            : tokenbt_init
 * @description		: 令牌桶初始化，初始化一个令牌桶对象
 * @param - cps     : 步长
 * @param - burst   : 上限零牌数
 * @return 			: 成功返回令牌桶对象，失败返回 NULL
 */
tokenbt_t *tokenbt_init(int cps, int burst){
    struct tokenbt_t* tb;
    int pos;
    pthread_once(&once_init, module_load);  // 任务函数只执行一次
    tb = malloc(sizeof(struct tokenbt_t));  // 分配空间
    if(tb == NULL){
        return NULL; //错误处理
    }
    tb->cps = cps;    //1s增加多少令牌数
    tb->burst = burst;//令牌数最大值
    tb->token = 0;    //初始令牌数
    pthread_mutex_init(&tb->mutex, NULL);
    pthread_cond_init(&tb->cond, NULL);
    //加锁，要处理tb里面的值了
    pthread_mutex_lock(&mutex_pool);
    pos = get_free_pos_unlocked();
    // 没有空余的空间了
    if (pos < 0)
    {
        pthread_mutex_unlock(&mutex_pool);      // 获取是失败别忘了解锁
        pthread_mutex_destroy(&tb->mutex);
        pthread_cond_destroy(&tb->cond);
        free(tb);
        syslog(LOG_ERR, "have not any pool pos...");
        // fprintf(stderr, "have not any pool pos...\n");
        return NULL;
    }
    //把tb插入到工作队之中
    tb->pos = pos;
    token_pool[pos] = tb;
    pthread_mutex_unlock(&mutex_pool);
    return tb;
}

/*
 * @name            : tokenbt_fetchtoken
 * @description		: 从令牌桶对象中取令牌
 * @param - token   : 令牌桶对象
 * @param - size    : 要取的零牌数
 * @return 			: 成功返回令牌，失败返回 -EINVAL
 */
int tokenbt_fetchtoken(tokenbt_t *token, int size){
    struct tokenbt_t *tb = (struct tokenbt_t *)token;
    int n;
    if (size <= 0 || token == NULL)     // 判断参数是否合法
        return -EINVAL;
    //每一次操作tb内部的东西都要加锁
    pthread_mutex_lock(&tb->mutex);
    while (tb->token <= 0){  //没有空余的token了，就等待
        pthread_cond_wait(&tb->cond, &tb->mutex);
    }
    n = tb->token;
    n = n < size ? n : size;        // 取得零牌数较小的那一个
    tb->token -= n;
    pthread_mutex_unlock(&tb->mutex);
    return n;   //返回的并不是size，而是实际取到的令牌数
}

/*
 * @name            : tokenbt_checktoken
 * @description		: 检查令牌桶对象的令牌数
 * @param - token   : 令牌桶对象
 * @return 			: 成功返回令牌，失败返回 -EINVAL
 */
int tokenbt_checktoken(tokenbt_t *token){
    struct tokenbt_t *tb = (struct tokenbt_t *)token;
    int token_size;
    if(tb == NULL){ //错误处理
        return -EINVAL;
    }
    pthread_mutex_lock(&tb->mutex);
    token_size = tb->token;
    pthread_mutex_unlock(&tb->mutex);
    return token_size;
}

/*
 * @name            : tokenbt_returntoken
 * @description		: 给令牌桶对象归还令牌
 * @param - token   : 令牌桶对象
 * @param - size    : 令牌
 * @return 			: 成功返回归还的令牌，失败返回 -EINVAL
 */
int tokenbt_returntoken(tokenbt_t *token, int size){
    struct tokenbt_t *tb = (struct tokenbt_t *)token;
    if (size <= 0 || token == NULL){ //参数正常
        return -EINVAL;
    }
    pthread_mutex_lock(&tb->mutex);
    tb->token += size;
    if (tb->token > tb->burst)
        tb->token = tb->burst;
    pthread_cond_broadcast(&tb->cond);
    pthread_mutex_unlock(&tb->mutex);
    return size;
}

/*
 * @name            : tokenbt_destroy
 * @description		: 销毁单个令牌桶对象
 * @param - token   : 令牌桶对象
 * @return 			: 成功返回 0，失败返回 -EINVAL
 */
int tokenbt_destroy(tokenbt_t *token){
    struct tokenbt_t *tb = (struct tokenbt_t *)token;
    if (token == NULL)
    return -EINVAL;
    pthread_mutex_lock(&mutex_pool);
    token_pool[tb->pos] = NULL;        //要把这个令牌桶从工作队列中拿出去
    pthread_mutex_unlock(&mutex_pool);
    // 然后再释放掉这个令牌桶内部的空间
    pthread_mutex_destroy(&tb->mutex);
    pthread_cond_destroy(&tb->cond);
    free(tb);
    tb = NULL;
    return 0;
}


/*
 * @name            : tokenbt_destroy_all
 * @description		: 销毁所有令牌桶对象
 * @param           : 无
 * @return 			: 成功返回 0
 */
int tokenbt_destroy_all(){
    for(int i=0;i<TOKENBUCKET_MAX;i++){
        if (token_pool[i] != NULL)
        {
            pthread_mutex_destroy(&token_pool[i]->mutex);
            pthread_cond_destroy(&token_pool[i]->cond);
        }
        free(token_pool[i]);
        token_pool[i] = NULL;
    }
    return 0;
}

/*
 * @name            : tokenbt_shutdown
 * @description		: 关闭令牌流控功能模块
 * @param           : 无
 * @return 			: 成功返回 0
 */
int tokenbt_shutdown()
{
    module_unload();
    return 0;
}

medialib

这个的任务就是将我们本地的medialib文件夹里面的内容转化为广播频道

主要完成两项任务：

1.节目单的组合(mlib_getchnlist函数)

2.读取频道内部的数据（mlib_readchn函数）

这两个函数是最关键的，之前所说的令牌桶就是在mlib_readchn函数里面用的，就是读取mp3文件的操作

首先我们来说一下结构

├── medialib
│   ├── ch1
│   │   ├── desc.txt
│   │   ├── 世界这么大还是遇见你.mp3
│   │   └── 起风了.mp3
│   ├── ch2
│   │   ├── desc.txt
│   │   ├── 带你去旅行.mp3
│   │   └── 再见只是陌生人.mp3
│   ├── ch3
│   │   ├── desc.txt
│   │   ├── Summertime Sadness.mp3
│   │   └── 学猫叫.mp3
│   └── ch4

每个ch之中都会有一个desc.txt的文件，这个里面就写的一些频道的基本的描述，比如:

ch1中内容就一行: pop music,起风了,世界这么大还是遇见你

所以我们要获取一下这个节目单的描述啥的还是很简单的，就是读取desc.txt文件就行了

我们看一下每个频道的管理的结构体

typedef struct channel_context_t {   //频道内容描述结构体
    chnid_t chnid;   // 频道ID
    char* desc;      // 频道描述
    glob_t globes;   // 目录项
    int pos;         // 当前歌曲在媒体库中的位置
    int fd;          // 当前歌曲的文件描述符
    off_t offset;    // 当前歌曲的读取偏移量
    tokenbt_t *tb;   // 流量控制
} channel_context_t;

// typedef struct {
//   size_t   gl_pathc;    /* 匹配到的路径数目 */
//   char   **gl_pathv;    /* 匹配到的路径名指针数组 */ 是一个指向匹配到的路径名字符串数组的指针。数组中的每个元素是一个路径名字符串，最后一个元素为 NULL，表示数组结束。
//   size_t   gl_offs;     /* 起始位置索引 */ 表示起始位置索引。如果设置了 GLOB_DOOFFS 标志，glob() 函数会在 gl_pathv 的开头预留 gl_offs 个空指针，实际的匹配路径名从第 gl_offs 个位置开始。
// } glob_t;

这里主要是几个参数可能不好理解

1.glob_t globes 这个主要是用来搜索一下*mp3的内容的，搜索出来的文件路径会存在结构体的gl_pathv里面

2.pos 就是因为会有很多.mp3文件，这个就标志着这是第几个文件，从0开始，这样的话方便列表循环播放

3.fd 这个就是目前打开的mp3的文件描述符，用以后面读文件

4.offset 这个文件读到哪里了，用这个来表示，方便下次继续读

5.tokenbt_t *tb; 令牌桶就用在这里

代码：

#ifndef __MEDIALIB_H__
#define __MEDIALIB_H__

#include <protocol.h>
#include <unistd.h>

//记录每一条节目单信息：频道号chnid，描述信息char* desc
typedef struct mlib_listdesc_t{
    chnid_t chnid;  // 频道唯一标识符，使用prot.h中定义的chnid_t类型
    char* desc;     // 频道描述信息，动态分配的字符串指针
}mlib_listdesc_t;

/**
 * 获取频道列表
 * @param  [输出参数] 接收频道信息结构体数组指针的地址
 * @param  [输出参数] 接收频道数量的指针
 * @return  成功返回0，失败返回-1
 */
int mlib_getchnlist(struct mlib_listdesc_t **,int *); // 从媒体库获取节目单信息和频道总个数

/**
 * 释放频道列表内存
 * @param  要释放的频道信息结构体数组指针
 * @return 总是返回0表示成功
 */
int mlib_freechnlist(struct mlib_listdesc_t *);       // 释放节目单信息存储所占的内存


/**
 * 释放chn_context数组的内存 
 * chn_context里面存储的是所有的广播频道号
 * @return 总是返回0表示成功
 */
int mlib_freechncontext();                            // 释放chn_context数组的内存


/*
 * @name            : mlib_readchn
 * @description		: 按频道读取对应媒体库流媒体内容
 * @param - chnid   : 频道号
 * @param - buf     : 存入流媒体内容的缓冲区指针
 * @param - size    : buf的最大容量
 * @return 			: 返回读取到的有效内容总长度
 */
ssize_t mlib_readchn(chnid_t, void *, size_t);        // 按频道读取对应媒体库流媒体内容


#endif // !__MEDIALIB_H__

#include <protocol.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <glob.h>
#include <syslog.h>
#include <sys/types.h>
#include <sys/stat.h>

#include "medialib.h"
#include "server_conf.h"
#include "tokenbucket.h"

#define PATHSIZE 4096      // 文件路径最大长度
#define NAMESIZE 256       // 文件名最大长度
#define LINEBUFSIZE 1024   // 读文件行缓冲区

#define MP3_BITRATE (128 * 1024)    // 128 * 1024 bps

typedef struct channel_context_t {   //频道内容描述结构体
    chnid_t chnid;   // 频道ID
    char* desc;      // 频道描述
    glob_t globes;   // 目录项
    int pos;         // 当前歌曲在媒体库中的位置
    int fd;          // 当前歌曲的文件描述符
    off_t offset;    // 当前歌曲的读取偏移量
    tokenbt_t *tb;   // 流量控制
} channel_context_t;

// typedef struct {
//   size_t   gl_pathc;    /* 匹配到的路径数目 */
//   char   **gl_pathv;    /* 匹配到的路径名指针数组 */ 是一个指向匹配到的路径名字符串数组的指针。数组中的每个元素是一个路径名字符串，最后一个元素为 NULL，表示数组结束。
//   size_t   gl_offs;     /* 起始位置索引 */ 表示起始位置索引。如果设置了 GLOB_DOOFFS 标志，glob() 函数会在 gl_pathv 的开头预留 gl_offs 个空指针，实际的匹配路径名从第 gl_offs 个位置开始。
// } glob_t;


static channel_context_t chn_context[MAXCHNID + 1];
static int total_chn = 0;           // 总共的有效频道个数


/*
 * @name            : getpathcontent
 * @description		: 从指定的路径中取得该频道所有需要的信息
 * @param - path    : 文件路径
 * @return 			: 成功返回 channel_context_t 对象地址; 失败返回 NULL
 */
static channel_context_t *getpathcontent(const char *path){
    char linebuf[LINEBUFSIZE];    //行缓冲
    char pathbuf[PATHSIZE];       //文件路径
    char namebuf[NAMESIZE];       //文件名

    int descfd, ret;
    channel_context_t *me;
    //由于是一个静态变量所以相当于一直在操作同一块内存 有叠加效果  静态频道ID计数器，初始为最小频道ID
    static int curr_chnid = MINCHNID;

    //先清空一下值
    memset(linebuf, 0, sizeof(linebuf));
    memset(pathbuf, 0, sizeof(pathbuf));
    memset(namebuf, 0, sizeof(namebuf));

    //赋值，字符串操作
    strncpy(pathbuf, path, PATHSIZE - 1);                         //path
    strncpy(namebuf, "/desc.txt", NAMESIZE - 1);                  //desc name
    strncat(pathbuf, namebuf, PATHSIZE - strlen(pathbuf) - 1);    //完整path

    descfd = open(pathbuf, O_RDONLY);
    if (descfd < 0){   //错误处理
        syslog(LOG_INFO, "%s is not a lib ...", pathbuf);
        return NULL;
    }

    ret = read(descfd, linebuf, LINEBUFSIZE);  //读文件
    if (ret == 0){ //错误处理
        syslog(LOG_INFO, "%s haven't anything ...", pathbuf);
        close(descfd);
        return NULL;
    }
    close(descfd);  //这里要注意哈，desc文件里面本来就只有一行
    // 申请空间
    me = malloc(sizeof(channel_context_t));
    if(me == NULL){ // 错误处理
        syslog(LOG_ERR, "malloc() : %s", strerror(errno));
        return NULL;
    }
    me->desc = strdup(linebuf); //strdup 复制字符串
    // 令牌桶 流量控制
    me->tb = tokenbt_init(MP3_BITRATE / 8, MP3_BITRATE / 8 * 5);
    if (me->tb == NULL){
        syslog(LOG_ERR, "tokenbt_init() : failed ...");
        free(me);
        return NULL;
    }
    // 这个时候已经把 desc.txt里的内容拿到了，现在该匹配mp3了
    memset(pathbuf, 0, sizeof(pathbuf));
    memset(namebuf, 0, sizeof(namebuf));
    // 一样的操作，path name path+name 只不过这里的是*.mp3
    strncpy(pathbuf, path, PATHSIZE - 1);
    strncpy(namebuf, "/*.mp3", NAMESIZE - 1);
    strncat(pathbuf, namebuf, PATHSIZE - strlen(pathbuf) - 1);

    ret = glob(pathbuf, 0, NULL, &me->globes);  //glob 匹配.mp3文件
    if (ret != 0){
        syslog(LOG_ERR, "glob() : failed ...");
        free(me);
        return NULL;
    }

    me->pos = 0;     //第1个pos，第一个mp3
    me->offset = 0;  //还没开始读，所以就offset（文件内部偏移）就是0

    //打开第一个mp3文件
    me->fd = open(me->globes.gl_pathv[me->pos], O_RDONLY);
    if (me->fd < 0){
        syslog(LOG_ERR, "open() : %s", strerror(errno));
        free(me);
        return NULL;
    }
    // 记录当前的 curr_chnid
    me->chnid = curr_chnid;
    curr_chnid++;
    return me;
}


/*
 * @name            : open_next
 * @description		: 打开指定频道的下一首歌曲
 * @param - chnid   : 频道号
 * @return 			: 成功返回 0; 失败返回 -1
 */
static int open_next(chnid_t chnid){
    int i;
    for (i = 0; i < chn_context[chnid].globes.gl_pathc; i++){
        chn_context[chnid].pos++;
        if (chn_context[chnid].pos == chn_context[chnid].globes.gl_pathc){
            chn_context[chnid].pos = 0;
        } // 列表循环
        close(chn_context[chnid].fd); //先关闭一下当前的
        // 打开下一首
        chn_context[chnid].fd = open(chn_context[chnid].globes.gl_pathv[chn_context[chnid].pos], O_RDONLY);
        if (chn_context[chnid].fd < 0){
            continue; //打不开倒不至于直接退出，再继续循环打开下一首
        }else{
            return 0;
        }
    }
    return -1;
}

/*
 * @name            : mlib_getchnlist
 * @description		: 从媒体库获取节目单信息和频道总个数
 * @param - list    : 传出参数，填入节目单信息
 * @param - size    : 传出参数，填入频道总个数
 * @return 			: 成功返回 0; 失败返回 -1
 */
int mlib_getchnlist(mlib_listdesc_t **list, int *size){
    int i,ret;
    glob_t globes;
    char path[PATHSIZE];
    channel_context_t *retmp;
    mlib_listdesc_t *tmp;

    memset(chn_context, 0, sizeof(chn_context));
    //初始化频道数组 
    for (i = MINCHNID; i < MAXCHNID + 1; i++){
        chn_context[i].chnid = -1; //设置为无效频道
    }
    //拿到medialib path  /*表示medialib下面所有的文件夹(ch1 ch2 ch3 ch4....)
    snprintf(path, PATHSIZE, "%s/*", server_conf.media_dir);

    ret = glob(path, 0, NULL, &globes);  //匹配一下
    if (ret != 0){
        syslog(LOG_ERR, "glob() : failed ...");
        return -1;
    }
    //分配一下空间
    tmp = malloc(sizeof(mlib_listdesc_t) * globes.gl_pathc);
    if (tmp == NULL){
        syslog(LOG_ERR, "malloc() : %s", strerror(errno));
        return -1;
    }

    for (i = 0; i < globes.gl_pathc; i++){       // 分别获取 ch1 ch2 ch3 ch4 中的频道内容并保存在 chn_context 中
        retmp = getpathcontent(globes.gl_pathv[i]);
        if (retmp != NULL){
            //注意这里会把retmp放到全局静态的chn_context之中
            memcpy(chn_context + retmp->chnid, retmp, sizeof(*retmp));
            tmp[total_chn].chnid = retmp->chnid;
            tmp[total_chn].desc = strdup(retmp->desc);
            total_chn++;
            free(retmp);
        }
    }
    *list = realloc(tmp, sizeof(mlib_listdesc_t) * total_chn);      // 给 *list 重新分配内存
    if (list == NULL){
        syslog(LOG_ERR, "realloc() : %s", strerror(errno));
        return -1;
    }
    *size = total_chn;
    globfree(&globes);
    return 0;
}

/*
 * @name            : mlib_freechnlist
 * @description		: 释放节目单信息存储所占的内存
 * @param - list    : 
 * @return 			: 成功返回 0
 */
int mlib_freechnlist(struct mlib_listdesc_t *list){
    int i;
    for (i = 0; i < total_chn; i++)
    {
        free(list[i].desc);
    }
    free(list);
    return 0;
}

/*
 * @name            : mlib_freechncontext
 * @description		: 释放chn_context数组的内存
 * @return 			: 成功返回 0
 */
int mlib_freechncontext(){
    int i;
    for (i = MINCHNID; i < MAXCHNID + 1; i++)
    {
        if (chn_context[i].chnid != -1)
        {
            free(chn_context[i].desc);
            globfree(&chn_context[i].globes);
            close(chn_context[i].fd);
        }
    }
    return 0;
}

/*
 * @name            : mlib_readchn
 * @description		: 按频道读取对应媒体库流媒体内容
 * @param - chnid   : 频道号
 * @param - buf     : 存入流媒体内容的缓冲区指针
 * @param - size    : buf的最大容量
 * @return 			: 返回读取到的有效内容总长度
 */
ssize_t mlib_readchn(chnid_t chnid, void *buf, size_t size){
    int token, len;
    //获取一下令牌桶之中的令牌，返回的是实际获取的令牌数量，因为可能不够size这么多令牌
    token = tokenbt_fetchtoken(chn_context[chnid].tb, size);
    while(1){
        //  从指定的偏移量处开始读取 read的话是从当前文件偏移量开始读取(当前文件偏移量存储再文件描述符之中)
        //  pread 1是可以指定偏移两 2是线程安全，不会引文其他线程读取了文件改变了文件偏移
        len = pread(chn_context[chnid].fd, buf, token, chn_context[chnid].offset);
        if (len < 0){
            if (errno == EINTR)
                return 0;
            syslog(LOG_ERR, "pread() : %s", strerror(errno));
            open_next(chnid);       // 如果这首歌曲读取失败了，那就切换下一首歌曲播放
        }else if(len == 0){
            syslog(LOG_INFO, "song: %s is over", chn_context[chnid].globes.gl_pathv[chn_context[chnid].pos]);
            open_next(chnid);       // 这首歌曲读取结束了，那就切换下一首歌曲播放
            break;
        }else{
            chn_context[chnid].offset += len;  //正常读取了就加offset
            break;
        }
    }
    if (len < token)                // 令牌没用完，归还令牌
    {
        tokenbt_returntoken(chn_context[chnid].tb, token - len);
    }
    return len;
}

list

之前不是medialib把咱么medialib文件夹里面的东西转化为频道了吗

这里就是组装一下，节目单频道，用来发送节目单的数据，代码很简单，看一下就好。

就是，拿到组装号的 节目点list，然后处理一下，然后发送就好。

这里的发送时调用的线程池（addtask）

就是一直发，1s发一次。另外频道号之前在protocol.h已经定义好了，频道列表的频道号固定是0

#ifndef __LIST_H__
#define __LIST_H__
#include "medialib.h"

int thr_list_create(mlib_listdesc_t *, int);   // 节目单频道任务创建

#endif // !__LIST_H__

#include <pthread.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <syslog.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/ip.h>

#include "threadpool.h"
#include "server_conf.h"
#include "list.h"

// 将需要发送的节目信息内容和内容的长度打包，传给线程池任务
typedef struct send_list_t
{
    int len;
    msg_list_t msg[0];
} send_list_t;

/*
 * @name            : sendlist
 * @description		: 定时发送节目单信息
 * @param - arg     : send_list_t类型
 * @param - shut    : 线程池当前开启状态
 * @return 			: 无
 */
void sendlist(void *arg, volatile int *shut){
    int len;
    send_list_t *info = (send_list_t *)arg;
    while (*shut == 0)
    {
        len = sendto(serversd, info->msg, info->len, 0, (void *)&sndaddr, sizeof(sndaddr));
        syslog(LOG_INFO, "%7s thread sendto %5d bytes, pool status is %d", "list", len, *shut);
        sleep(1);
    }
}

/*
 * @name            : thr_list_create
 * @description		: 节目单频道任务创建
 * @param - list    : 从媒体库读取的原始节目信息数据
 * @param - size    : 频道总个数
 * @return 			: 成功返回 0; 失败返回 -1
 */
int thr_list_create(mlib_listdesc_t *list, int size){
    int len, totalsize;
    int i;
    msg_list_t *msg_list;
    msg_listdesc_t *desc_list;
    send_list_t *info;

    totalsize = sizeof(chnid_t);
    for (i = 0; i < size; i++){
        totalsize += sizeof(msg_listdesc_t) + strlen(list[i].desc);     // 统计节目单信息有效数据的总长度
    }

    info = malloc(totalsize + sizeof(int)); // 1节目单1 这样的 所以最前面得加个int
    if (info == NULL){
        syslog(LOG_ERR, "malloc() : %s", strerror(errno));
        return -1;
    }
    memset(info, 0, totalsize + sizeof(int));
    info->len = totalsize;              // 将有效数据的总长度填入申请的内存中
    msg_list = info->msg;

    msg_list->chnid = LISTCHNID;        // 填入 LISTCHNID 频道号 0 号只广播频道信息
    desc_list = msg_list->list;


    for (i = 0; i < size; i++)          // 将节目单信息的有效数据内容填入申请的内存中
    {
        len = sizeof(msg_listdesc_t) + strlen(list[i].desc);
        desc_list->chnid = list[i].chnid;
        desc_list->deslength = htons(len);
        strncpy((void *)desc_list->desc, list[i].desc, strlen(list[i].desc));
        desc_list = (void *)(((char *)desc_list) + len);
    }
    threadpool_addtask(pool, sendlist, info); // 向任务队列添加一个任务
    return 0;
}

channel

这个就是某个频道，一直发送数据了，不管有没有客户端都要一直发送

sendchannel里面调用了mlib_readchn，发送完一个mp3文件之后会列表循环，不会停止发送哈

之前也说了mlib_readchn里面使用了令牌桶进行了流量控制

#ifndef __CHANNEL_H__
#define __CHANNEL_H__

#include <protocol.h>

int thr_channel_create(chnid_t);    // 流媒体音乐频道任务创建

#endif // !__CHANNEL_H__

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <syslog.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/ip.h>

#include "channel.h"
#include "server_conf.h"
#include "medialib.h"

/*
 * @name            : sendchannel
 * @description		: 读取媒体库中流媒体的内容并发送
 * @param - arg     : msg_channel_t类型
 * @param - shut    : 线程池当前开启状态
 * @return 			: 无
 */
static void sendchannel(void *arg, volatile int *shut){
    msg_channel_t *context = (msg_channel_t *)arg;
    int len;
    while (*shut == 0){
        memset(context->data, 0, MAX_CHANNEL_DATA - sizeof(chnid_t));
        //读取频道里面的内容
        len = mlib_readchn(context->chnid, context->data, MAX_CHANNEL_DATA - sizeof(chnid_t));
        len = sendto(serversd, context, len + sizeof(chnid_t), 0, (void *)&sndaddr, sizeof(sndaddr));
        syslog(LOG_INFO, "%7s thread sendto %5d bytes, pool status is %d", "channel", len, *shut);
    }
}

/*
 * @name            : thr_channel_create
 * @description		: 流媒体音乐频道任务创建
 * @param - chnid   : 频道号
 * @return 			: 成功返回 0; 失败返回 -1
 */
int thr_channel_create(chnid_t chnid){
    msg_channel_t *context;
    context = malloc(MAX_CHANNEL_DATA);     // 申请内存，将本块内存地址传入线程池中
    if (context == NULL){
        syslog(LOG_ERR, "malloc() : %s", strerror(errno));
        // fprintf(stderr, "malloc() : %s\n", strerror(errno));
        return -1;
    }
    memset(context, 0, MAX_CHANNEL_DATA);
    context->chnid = chnid;
    threadpool_addtask(pool, sendchannel, context);     // context指向的内存由线程池负责释放
    return 0;
}

server

前面都讲了，这里就没啥好讲的了，就是将所有的东西组合一下就好

两点注意：

1.守护进程

就是将整个的进程置于后台，终端就不输出了

就是fork一个子进程，然后脱离父进程，使用子进程来执行代码

将输出重定向到 /dev/null 里面，/dev/null是一个linux内置的一个设备，往里输出的话就表示，我不需要这些输出，直接丢弃

setsid 创建新会话，脱离终端，即使终端关闭了也不会断开

static int daemon_init(){
    pid_t pid;
    int fd;
    pid = fork();
    if(pid < 0){
        syslog(LOG_ERR, "fork() : %s", strerror(errno));
        return -1;
    }else if(pid > 0){ //退出父进程
        //确保守护进程脱离父进程控制，成为后台进程。
        exit(EXIT_SUCCESS);
    }

    // dev/null 这是一个特殊的设备，输入进去之后，所有的东西都会被丢弃
    fd = open("/dev/null", O_RDWR);
    if(fd < 0){
        syslog(LOG_ERR, "open() : %s\n", strerror(errno));
        return -1;
    }
    // 守护进程的输入/输出/错误不再关联终端，避免干扰前台。
    dup2(fd, STDIN_FILENO);   // 标准输入
    dup2(fd, STDOUT_FILENO);  // 标准输出
    dup2(fd, STDERR_FILENO);  // 标准错误

    if (fd > STDERR_FILENO){
        close(fd);
    }

    chdir("/");    // 更改工作目录到根目录
    // 设置文件创建时的默认权限掩码。umask(0) 表示新文件权限为 0777 & ~0 = 0777（即 rwxrwxrwx）
    // 确保守护进程创建的文件/目录具有最大权限，由程序自行控制具体权限。
    umask(0);      // 重置文件权限掩码
    // 创建一个新的会话（Session），并成为会话的首进程。
    // 确保守护进程没有控制终端（TTY），即使终端关闭，守护进程仍可运行。
    setsid();      // 创建新会话，脱离终端控制

    return 0;
}

后面就没啥了，就是socket的创建以及各个函数的调用，这个看代码一看就明白

#include <protocol.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <syslog.h>
#include <fcntl.h>
#include <error.h>
#include <errno.h>
#include <string.h>
#include <signal.h>
#include <getopt.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <netinet/ip.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <net/if.h>

#include "server_conf.h"
#include "threadpool.h"
#include "medialib.h"
#include "tokenbucket.h"
#include "channel.h"

#include "list.h"

int serversd;                 //全局性变量，在server_conf里面有声明 服务端套接字
ThreadPool_t *pool;           //全局性变量，在server_conf里面有声明 线程池
struct sockaddr_in sndaddr;   //全局性变量，在server_conf里面有声明 发送目的地址
static mlib_listdesc_t *list; //节目单

//全局性变量，在server_conf里面有声明 服务端的一些配置
server_conf_t server_conf =
    {
        .media_dir  = DEFAULT_MEDIADIR,
        .rcvport    = DEFAULT_RECVPORT,
        .runmode    = RUN_FOREGROUND,
        .ifname     = DEFAULT_IF,
        .mgroup     = DEFAULT_MGROUP
    };

// 命令行参数
struct option opt[] =
    {
        {"mgroup"  , required_argument, NULL, 'M'},
        {"port"    , required_argument, NULL, 'P'},
        {"mediadir", required_argument, NULL, 'D'},
        {"runmode" , required_argument, NULL, 'R'},
        {"ifname"  , required_argument, NULL, 'I'},
        {"help"    , no_argument      , NULL, 'H'}
    };

// 命令行参数帮助
static void print_help()
{
    printf("-M --mgroup     自定义多播组地址\n");
    printf("-P --port       自定义发送端口  \n");
    printf("-D --mediadir   自定义媒体库路径\n");
    printf("-R --runmode    自定义运行模式  \n");
    printf("-I --ifname     自定义网卡名称  \n");
    printf("-H --help       显示帮助       \n");
}

// 守护进程
static int daemon_init(){
    pid_t pid;
    int fd;
    pid = fork();
    if(pid < 0){
        syslog(LOG_ERR, "fork() : %s", strerror(errno));
        return -1;
    }else if(pid > 0){ //退出父进程
        //确保守护进程脱离父进程控制，成为后台进程。
        exit(EXIT_SUCCESS);
    }

    // dev/null 这是一个特殊的设备，输入进去之后，所有的东西都会被丢弃
    fd = open("/dev/null", O_RDWR);
    if(fd < 0){
        syslog(LOG_ERR, "open() : %s\n", strerror(errno));
        return -1;
    }
    // 守护进程的输入/输出/错误不再关联终端，避免干扰前台。
    dup2(fd, STDIN_FILENO);   // 标准输入
    dup2(fd, STDOUT_FILENO);  // 标准输出
    dup2(fd, STDERR_FILENO);  // 标准错误

    if (fd > STDERR_FILENO){
        close(fd);
    }

    chdir("/");    // 更改工作目录到根目录
    // 设置文件创建时的默认权限掩码。umask(0) 表示新文件权限为 0777 & ~0 = 0777（即 rwxrwxrwx）
    // 确保守护进程创建的文件/目录具有最大权限，由程序自行控制具体权限。
    umask(0);      // 重置文件权限掩码
    // 创建一个新的会话（Session），并成为会话的首进程。
    // 确保守护进程没有控制终端（TTY），即使终端关闭，守护进程仍可运行。
    setsid();      // 创建新会话，脱离终端控制

    return 0;
}

/* 信号处理函数：清理资源并退出 */
static void daemon_exit(int s) {
    threadpool_destroy(pool);     // 销毁线程池
    mlib_freechnlist(list);       // 释放频道列表内存
    mlib_freechncontext();        // 释放频道上下文资源
    tokenbt_shutdown();           // 关闭令牌桶
    close(serversd);              // 关闭套接字
    closelog();                   // 关闭系统日志
    exit(EXIT_SUCCESS);           // 正常退出
}

/* 初始化UDP套接字 */
static int socket_init() {
    int ret;
    struct ip_mreqn mreq; // 多播请求结构体
    /*struct ip_mreqn：用于配置多播组的参数，包含以下字段：
        imr_multiaddr：多播组 IP 地址。
        imr_address：本地接口 IP 地址。
        imr_ifindex：网卡接口的索引。 
    */
    // 创建UDP套接字
    serversd = socket(AF_INET, SOCK_DGRAM, 0);
    if (serversd < 0) {
        syslog(LOG_ERR, "socket() : %s", strerror(errno));
        exit(EXIT_FAILURE);
    }

    // 设置多播组参数
    inet_pton(AF_INET, server_conf.mgroup, &mreq.imr_multiaddr); // 多播组地址
    inet_pton(AF_INET, "0.0.0.0", &mreq.imr_address);           // 本地任意地址
    mreq.imr_ifindex = if_nametoindex(server_conf.ifname);     // 网卡名称转索引
    /*
    inet_pton：将字符串形式的 IP 地址转换为二进制格式。
    server_conf.mgroup：配置中的多播组地址（如 "239.0.0.1"）。
    "0.0.0.0"：表示绑定到本地所有可用接口。
    if_nametoindex：将网卡名称（如 "eth0"）转换为系统索引。
    若网卡不存在，此函数会失败，但代码中未处理该错误。
     */

     // 告诉内核通过指定的网卡（server_conf.ifname）发送多播数据。
    ret = setsockopt(serversd, IPPROTO_IP, IP_MULTICAST_IF, &mreq, sizeof(mreq));
    if (ret < 0)
    {
        syslog(LOG_ERR, "setsockopt() : %s", strerror(errno));
        exit(EXIT_FAILURE);
    }

    sndaddr.sin_family = AF_INET;
    sndaddr.sin_port = htons(atoi(server_conf.rcvport));
    inet_pton(AF_INET, server_conf.mgroup, &sndaddr.sin_addr);

    return 0;
}

int main(int argc, char **argv) {
    int i, ret;
    int arg;
    int list_size;                // 频道列表大小
    struct sigaction action;      // 信号处理结构体

    // 打开系统日志，标识为"netradio"，包含PID和输出到stderr
    openlog("netradio", LOG_PID | LOG_PERROR, LOG_DAEMON);

    while (1) {
        arg = getopt_long(argc, argv, "M:P:D:R:I:H", opt, NULL);
        if (arg == -1) //没有参数就都是默认的，不用解析
            break;
        switch (arg) {
        case 'M':
            server_conf.mgroup = optarg; // 设置多播组地址
            break;
        case 'P':
            server_conf.rcvport = optarg; // 设置接收端口
            break;
        case 'D':
            server_conf.media_dir = optarg; // 设置媒体目录
            break;
        case 'R':
            // 设置运行模式（0:前台，1:守护进程）
            if (atoi(optarg) == 1 || atoi(optarg) == 0) {
                server_conf.runmode = (enum RNUMODE)atoi(optarg);
            } else {
                syslog(LOG_ERR, "参数错误！详见");
                print_help();
                exit(EXIT_FAILURE);
            }
            break;
        case 'I':
            server_conf.ifname = optarg; // 设置网卡名称
            break;
        case 'H':
            print_help(); // 显示帮助
            exit(EXIT_SUCCESS);
        default:
            syslog(LOG_ERR, "参数错误！详见");
            print_help();
            exit(EXIT_FAILURE);
        }
    }
    syslog(LOG_INFO, "当前配置：\n多播组IP：\t%s\n端口：\t\t%s\n媒体库路径：\t%s\n运行模式：\t%d\n网卡名：\t%s\n",
        server_conf.mgroup,
        server_conf.rcvport,
        server_conf.media_dir,
        server_conf.runmode,
        server_conf.ifname);
    // 判断运行模式，如果是守护进程模式则初始化
    if (server_conf.runmode == RUN_DAEMON)  // 判断系统运行模式
    {
        ret = daemon_init();
        if (ret < 0){
            syslog(LOG_ERR, "daemon_init() failed ...");
            exit(EXIT_FAILURE);
        }
    }

    // 信号处理
    action.sa_flags = 0;
    sigemptyset(&action.sa_mask);    // 清空信号掩码
    sigaddset(&action.sa_mask, SIGINT);  // Ctrl+C
    sigaddset(&action.sa_mask, SIGQUIT); // Ctrl+\ //
    sigaddset(&action.sa_mask, SIGTSTP); // Ctrl+Z
    action.sa_handler = daemon_exit;      // 绑定处理函数 //都是退出
    sigaction(SIGINT, &action, NULL); // 注册信号捕捉函数
    sigaction(SIGQUIT, &action, NULL);
    sigaction(SIGTSTP, &action, NULL);

    // 初始化套接字
    socket_init();

    // 创建线程池（核心5线程，最大20线程，队列容量20）
    pool = threadpool_create(5, 20, 20);
    if (pool == NULL) {
        syslog(LOG_ERR, "threadpool_create() : failed ...");
        exit(EXIT_FAILURE);
    }

    // 获取媒体库频道列表
    ret = mlib_getchnlist(&list, &list_size);
    if (ret < 0) {
        syslog(LOG_ERR, "mlib_getchnlist() : failed ...");
        exit(EXIT_FAILURE);
    }

    // 创建频道列表线程
    ret = thr_list_create(list, list_size);
    if (ret < 0) {
        syslog(LOG_ERR, "thr_list_create() : failed ...");
        exit(EXIT_FAILURE);
    }

    // 为每个频道创建发送线程
    for (i = 0; i < list_size; i++) {
        ret = thr_channel_create(list[i].chnid);
        if (ret < 0) {
            syslog(LOG_ERR, "thr_channel_create() : failed ...");
            exit(EXIT_FAILURE);
        }
    }

    // 主循环挂起，等待信号
    while (1)
        pause();  // 阻塞主线程，等待信号 这里要注意哈，为舍么要阻塞？因为ie前面发送节目单信息以及每个频道里面的信息都申请了 额外的线程来做了，这里就没必要再做了

    exit(EXIT_SUCCESS);
}

客户端

任务就是接收数据，两个进程，父进程进行数据的接收，输入到管道里面，然后子进程拿数据输入到ffmpeg里面来播放，就是这个流程, 基本没有啥坑

唯一需要注意的点就是如何选择自己的频道

首先要明白一点，加入到组播里面之后，client会接收所有的数据

因此需要先进行判断，首先得先拿到节目单，也就是ID为0的包。不是ID为0的包就丢弃。

然后把包里面的内容输出到终端让客户来选择那个频道，比如输入1，那么就选择1频道

后面拿到包之后先对比一下是不是1频道，是的话就拿过来，处理。不是的话就丢弃就好。

其他的就没啥的，下面是代码

#ifndef __CLIENT_H__
#define __CLIENT_H__

// #define DEFAULT_PALYERCMD "/usr/bin/mplayer -"     使用mpg123播放器并将输出重定向到/dev/null（静默模式）
// /dev/null 是一个特殊的文件，通常被称为“空设备”或“位桶”（bit bucket） 
// 任何写入 /dev/null 的内容都会被永久丢弃，不会保存在任何地方。同时，从 /dev/null 读取内容时，总是会返回空（EOF，即文件结束符）。
#define DEFAULT_PALYERCMD "/usr/bin/mpg123 - > /dev/null" //mpg123播放mp3文件，并且把一些输出给丢弃掉 

#include <stdint.h>

typedef struct client_conf_t
{
    char *mgroup;       // 多播组IP地址
    char *revport;      // 接收端口
    char *playercmd;    // 播放器命令
} client_conf_t;

#endif // !__CLIENT_H__

#include <protocol.h>

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <getopt.h>
#include <net/if.h>
#include <errno.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <arpa/inet.h>

#include "client.h"

// 定义全局变量，程序退出时释放内存。 
msg_list_t *msg_list = NULL;        // 节目单数据缓冲区指针
msg_channel_t *msg_channel = NULL;  // 频道数据缓冲区指针
int sfd;                            // 套接字文件描述符

client_conf_t conf = // client 配置
    {
        .mgroup = DEFAULT_MGROUP,         // 默认多播组地址
        .revport = DEFAULT_RECVPORT,      // 默认接收端口
        .playercmd = DEFAULT_PALYERCMD};  // 默认播放器命令

// 命令行参数解析
struct option opt[] =
    {
        {"mgroup", required_argument, NULL, 'M'},       // 多播组地址选项
        {"port", required_argument, NULL, 'P'},         // 接收端口选项
        {"player", required_argument, NULL, 'p'},       // 播放器命令选项
        {"help", no_argument, NULL, 'H'}};              // 帮助选项

// 命令行参数帮助 // 打印帮助信息
static void print_help()
{
    printf("-M --mgroup 自定义多播组地址\n");
    printf("-P --port   自定义接收端口  \n");
    printf("-p --player 自定义音乐解码器\n");
    printf("-H --help   显示帮助       \n");
}

/*
 * @name            : writen
 * @description		: 自定义封装函数，保证写足 count 字节
 * @param - fd 	    : 文件描述符
 * @param - buf 	: 要写入的内容
 * @param - count 	: 要写入的内容总长度
 * @return 			: 成功返回写入的字节数; 失败返回 -1
 */
static ssize_t writen(int fd, const void *buf, size_t count){
    size_t len, total, ret;
    total = count;
    for (len = 0; total > 0; len += ret, total -= ret){
    again:
        ret = write(fd, buf + len, total);
        if (ret < 0)
        {
            if (errno == EINTR) // 中断系统调用，重启 write
                goto again;
            fprintf(stderr, "write() : %s\n", strerror(errno));
            return -1;
        }
    }
    return len;

}

/*
 * @name            : exit_action
 * @description		: 信号捕捉函数，用于退出前清理
 * @param - s 	    : 信号
 * @return 			: 无
 */
static void exit_action(int s){
    pid_t pid;
    pid = getpgid(getpid());  // 获取进程组ID
    // 释放动态分配的内存
    if (msg_list != NULL)
        free(msg_list);
    if (msg_channel != NULL)
        free(msg_channel);
    close(sfd);           // 关闭套接字
    kill(-pid, SIGQUIT);  // 向进程组发送退出信号
    fprintf(stdout, "\nthis programme is going to exit...\n");
    exit(EXIT_SUCCESS);
}

int main(int argc, char **argv){

    int arg;       // 临时存储getopt返回值
    int ret;       // 通用返回值存储
    int len;       // 接收数据长度
    int val;       // 临时整数值存储
    int chosen;    // 用户选择的频道ID
    int fd[2];     // 管道文件描述符数组
    char ip[20];   // 存储IP字符串
    pid_t pid;     // 进程ID
    socklen_t socklen;   // 套接字地址结构长度
    uint64_t receive_buf_size = 20 * 1024 * 1024; // 20MB  // 接收缓冲区大小（20MB）
    struct ip_mreqn group;      // 多播组结构体
    struct sockaddr_in addr,    // 本地绑定地址
                    list_addr,  // 节目单来源地址
                    data_addr;  // 数据来源地址
    struct sigaction action;    // 信号处理结构体

    while (1)
    {
        arg = getopt_long(argc, argv, "P:M:p:H", opt, NULL); // 命令行参数解析
        if (arg == -1)
            break;
        switch (arg)
        {
        case 'P':
            conf.revport = optarg;      // 设置接收端口
            break;
        case 'M':
            conf.mgroup = optarg;       // 设置多播组地址
            break;
        case 'p':
            conf.playercmd = optarg;    // 设置播放器命令
            break;
        case 'H':                       // 显示帮助
            print_help();
            break;
        default:
            fprintf(stderr, "参数错误！详见\n");
            print_help();
            exit(EXIT_FAILURE);
            break;
        }
    }
    // 打印当前配置
    fprintf(stdout, "当前配置：\n多播组IP:\t%s\n端口：\t\t%s\n播放器：\t%s\n",
            conf.mgroup, conf.revport, conf.playercmd);

    ret = pipe(fd); // 创建匿名管道创建管道用于与播放器通信
    if (ret < 0){
        fprintf(stderr, "pipe() : %s\n", strerror(errno));
        exit(EXIT_FAILURE);
    }

    pid = fork(); // 创建子进程 // 创建子进程用于运行播放器
    if (pid < 0){  //创建失败
        fprintf(stderr, "fork() : %s\n", strerror(errno));
        exit(EXIT_FAILURE);
    }else if(pid == 0){  //子进程负责将父进程读取到的数据通过管道馈入到ffmpeg里面播放
        close(fd[1]);                 // 关闭写端（子进程只读）
        dup2(fd[0], STDIN_FILENO);    // 将管道读端重定向到标准输入
        close(fd[0]);                 // 关闭原始读端
        execl("/bin/sh", "sh", "-c", conf.playercmd, NULL); // 使用shell解释器来运行 mpg123，子进程被替换成mpg123
        fprintf(stderr, "execl() : %s\n", strerror(errno));
        exit(EXIT_FAILURE);
    }
    // 父进程
    close(fd[0]); //父进程不需要读管道
    // 配置信号处理结构体
    action.sa_flags = 0;                   // 不使用特殊标志
    sigemptyset(&action.sa_mask);          // 清空信号屏蔽字
    // 添加要阻塞的信号（处理期间屏蔽这些信号）
    sigaddset(&action.sa_mask, SIGINT);    // Ctrl+C
    sigaddset(&action.sa_mask, SIGQUIT);   // Ctrl+\ //
    sigaddset(&action.sa_mask, SIGTSTP);   // Ctrl+Z
    action.sa_handler = exit_action;       // 设置统一信号处理函数
    // 注册信号处理器
    sigaction(SIGINT, &action, NULL);   // 注册Ctrl+C处理
    sigaction(SIGQUIT, &action, NULL);  // 注册退出处理
    sigaction(SIGTSTP, &action, NULL);  // 注册暂停处理

    // 创建UDP套接字
    sfd = socket(AF_INET, SOCK_DGRAM, 0);
    // 设置绑定地址结构
    addr.sin_family = AF_INET;
    inet_pton(AF_INET, "0.0.0.0", &addr.sin_addr);
    addr.sin_port = htons(atoi(conf.revport));
    // 绑定套接字
    ret = bind(sfd, (void *)&addr, sizeof(addr)); // 绑定本地 IP ，端口

    if (ret < 0){
        fprintf(stderr, "bind() : %s\n", strerror(errno));
        exit(EXIT_FAILURE);
    }

    // 设置接收缓冲区大小
    ret = setsockopt(sfd, SOL_SOCKET, SO_RCVBUF, &receive_buf_size, sizeof(receive_buf_size)); // 设置套接字接收缓冲区 20 MB
    if (ret < 0)
    {
        fprintf(stderr, "SO_RCVBUF : %s\n", strerror(errno));
        exit(EXIT_FAILURE);
    }

    // 允许组播数据回环（本机接收自己发送的数据）
    val = 1;
    ret = setsockopt(sfd, IPPROTO_IP, IP_MULTICAST_LOOP, &(val), sizeof(val)); // 允许组播数据包本地回环
    if (ret < 0)
    {
        fprintf(stderr, "IP_MULTICAST_LOOP : %s\n", strerror(errno));
        exit(EXIT_FAILURE);
    }

        // 设置多播组参数
    inet_pton(AF_INET, conf.mgroup, &group.imr_multiaddr);       // 多播组地址
    inet_pton(AF_INET, "0.0.0.0", &group.imr_address);           // 本地接口地址
    group.imr_ifindex = if_nametoindex("ens33");                                 // 绑定自己的网卡
    ret = setsockopt(sfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &group, sizeof(group)); // 加入多播组

    if (ret < 0)
    {
        fprintf(stderr, "IP_ADD_MEMBERSHIP() : %s\n", strerror(errno));
        exit(EXIT_FAILURE);
    }
    // 分配节目单数据缓冲区
    msg_list = malloc(MAX_LISTCHN_DATA);
    if (msg_list == NULL){ //错误处理
        fprintf(stderr, "malloc() : %s\n", strerror(errno));
    }
    socklen = sizeof(struct sockaddr_in);

    while(1){
        memset(msg_list, 0, MAX_LISTCHN_DATA);    // 清空缓冲区
        len = recvfrom(sfd, msg_list, MAX_LISTCHN_DATA, 0, (void *)&list_addr, &socklen); // 接收节目单包
        if (len < sizeof(msg_list_t))
        {
            fprintf(stderr, "data is too short, len = %d...\n", len);
            continue;
        }
         // 检查是否为节目单数据（根据预设的LISTCHNID判断）
        if (msg_list->chnid == LISTCHNID) // 如果是节目单包则保留，不是则丢弃
        {
            fprintf(stdout, "list from IP = %s, Port = %d\n",
                    inet_ntop(AF_INET, &list_addr.sin_addr, ip, sizeof(ip)),
                    ntohs(list_addr.sin_port));
            break;
        }
    }

    // 解析并打印节目单信息
    msg_listdesc_t *desc;
    for (desc = msg_list->list; (char *)desc < (char *)msg_list + len; desc = (void *)((char *)desc + ntohs(desc->deslength)))
    {
        fprintf(stdout, "chnid = %d, description = %s\n", desc->chnid, desc->desc);
    }
    free(msg_list); // 释放节目单缓冲区
    msg_list = NULL;

    // 用户选择频道
    fprintf(stdout, "请输入收听的频道号码，按回车结束！\n");
    while (1)
    {
        fflush(NULL);
        ret = scanf("%d", &chosen);
        if (ret != 1)
            exit(EXIT_FAILURE);
        else if (ret == 1)
            break;
    }

    // 分配频道数据缓冲区
    msg_channel = malloc(MAX_CHANNEL_DATA);
    if (msg_channel == NULL)
    {
        fprintf(stderr, "malloc() : %s\n", strerror(errno));
    }

    // 持续接收数据
    socklen = sizeof(struct sockaddr_in);

    while(1){
        memset(msg_channel, 0, MAX_CHANNEL_DATA);     // 清空缓冲区
        len = recvfrom(sfd, msg_channel, MAX_CHANNEL_DATA, 0, (void *)&data_addr, &socklen); // 接收频道内容包
        if (len < sizeof(msg_channel_t)){ // UDP不是面向字节流的，接收的话就是一个包
            fprintf(stderr, "data is too short, len = %d...\n", len);
            continue;
        }else if(data_addr.sin_addr.s_addr != list_addr.sin_addr.s_addr || data_addr.sin_port != list_addr.sin_port){
        // 验证数据包和节目单数据包是否为同一服务端发送，防止干扰
            fprintf(stderr, "data is not match!\n");
            continue;
        }

        if (msg_channel->chnid == chosen)
        {
            fprintf(stdout, "recv %d length data!\n", len);
            ret = writen(fd[1], msg_channel->data, len - sizeof(msg_channel->chnid)); // 写入管道
            if (ret < 0){
                exit(EXIT_FAILURE);
            }
        }
    }
    exit(EXIT_SUCCESS);
}

总结

感觉比webserver还要简单一些，就是这个里面的结构体各种定义很容易迷，这就是不如C++ class封装之处了。整体来看还是很值的做的。