嵌入式流媒体消息分发队列设计

      在流媒体系统设计中，经常会涉及多线程间的通信问题，需要相互分发消费一些信息以及实时音视频数据，如何才能高效安全的简洁的实现这一目标呢， 可以采用ringbuffer，也可以采用链表等算法。我在实践中做了一个基于信号量和固定指针数组的方案，实现了简单的数据发布订阅，pMessage可以自定义为一个适配自己的消息传递数据结构指针，用于传递消息，如数据指向的指针，数据类型，时间戳，数据长度，或者信令，在嵌入式系统中，采用预分配的全局数组，可以减少动态分配带来的风险和时间波动。通过设置一些信号量及互斥来实现系统级的线程调度，为了使使用更简洁，封装了一些简单的api，实现音频信息 视频信息以及信令信息的接口，将具体的逻辑封装起来，线程中直接调用这些接口就可以实现系统级的订阅等待和数据发布功能，当然这个代码不一定设计最科学和合理，但的确为我的系统带来了多线程处理过程的简洁明快的好处，也在实践中运行良好，现将他开源出来，也希望各位大佬指正，帮助我提高。好了，各位有啥好的意见和建议欢迎跟我交流。本代码用于我的嵌入式webrtc系统，大大简化了我的系统的复杂度。

#include <stdio.h> 
#include <pthread.h> 
#include<semaphore.h>
#include "common.h"
#define BUFFER_SIZE 50 // 缓冲区数量 
struct prodcons 
{ 
	// 缓冲区相关数据结构 
	pMessage buffer[BUFFER_SIZE]; /* 实际数据存放的数组*/ 
	pthread_mutex_t lock; /* 互斥体lock ⽤于对缓冲区的互斥操作 */ 
	int readpos, writepos; /* 读写指针*/ 
	pthread_cond_t notempty; /* 缓冲区⾮空的条件变量 */ 
	pthread_cond_t notfull; /* 缓冲区未满的条件变量 */ 
}; 
#define OVER ( - 1) 
struct prodcons pmsgbuffer; 
struct prodcons pvideo_streambuffer; 
struct prodcons paudio_streambuffer; 
struct prodcons pplayvideo_streambuffer; 
struct prodcons pplayaudio_streambuffer; 
 
sem_t sem_answerwait, sem_offerwait,sem_videostreamwait,sem_audiostreamwait,sem_playvideostreamwait,sem_playaudiostreamwait;
/* 初始化缓冲区结构 */ 
void init(struct prodcons *b) 
{ 
	pthread_mutex_init(&b->lock, NULL); 
	pthread_cond_init(&b->notempty, NULL); 
	pthread_cond_init(&b->notfull, NULL); 
	b->readpos = 0; 
	b->writepos = 0; 
} 
/* 将产品放⼊缓冲区,这⾥是存⼊⼀个整数*/ 
void put(struct prodcons *b, pMessage data) 
{ 
	pthread_mutex_lock(&b->lock); 
	/* 等待缓冲区未满*/ 
	if ((b->writepos + 1) % BUFFER_SIZE == b->readpos) 
	{ 
	  pthread_cond_wait(&b->notfull, &b->lock); 
	} 
	/* 写数据,并移动指针 */ 
	b->buffer[b->writepos] = data; 
	b->writepos++; 
	if (b->writepos >= BUFFER_SIZE) 
	  b->writepos = 0; 
	/* 设置缓冲区⾮空的条件变量*/ 
	pthread_cond_signal(&b->notempty); 
	pthread_mutex_unlock(&b->lock); 
}  
/* 从缓冲区中取出整数*/ 
pMessage get(struct prodcons *b) { 
	pMessage data; 
	pthread_mutex_lock(&b->lock); 
	/* 等待缓冲区⾮空*/ 
	if (b->writepos == b->readpos) 
	{ 
	  pthread_cond_wait(&b->notempty, &b->lock); 
	} 
	/* 读数据,移动读指针*/ 
	data = b->buffer[b->readpos]; 
	b->readpos++; 
	if (b->readpos >= BUFFER_SIZE) 
	   b->readpos = 0; 
	/* 设置缓冲区未满的条件变量*/
	pthread_cond_signal(&b->notfull); 
	pthread_mutex_unlock(&b->lock); 
	return data; 
} 
/* 将产品放⼊缓冲区,这⾥是存⼊⼀个整数*/ 
void putstream(struct prodcons *b, ptstreamInfo data) 
{ 
	pthread_mutex_lock(&b->lock); 
	/* 等待缓冲区未满*/ 
	if ((b->writepos + 1) % BUFFER_SIZE == b->readpos) 
	{ 
	  pthread_cond_wait(&b->notfull, &b->lock); 
	} 
	/* 写数据,并移动指针 */ 
	b->buffer[b->writepos] = data; 
	b->writepos++; 
	if (b->writepos >= BUFFER_SIZE) 
	  b->writepos = 0; 
	/* 设置缓冲区⾮空的条件变量*/ 
	pthread_cond_signal(&b->notempty); 
	pthread_mutex_unlock(&b->lock); 
}  
/* 从缓冲区中取出整数*/ 
ptstreamInfo getstream(struct prodcons *b) { 
	ptstreamInfo data; 
	pthread_mutex_lock(&b->lock); 
	/* 等待缓冲区⾮空*/ 
	if (b->writepos == b->readpos) 
	{ 
	  pthread_cond_wait(&b->notempty, &b->lock); 
	} 
	/* 读数据,移动读指针*/ 
	data = b->buffer[b->readpos]; 
	b->readpos++; 
	if (b->readpos >= BUFFER_SIZE) 
	   b->readpos = 0; 
	/* 设置缓冲区未满的条件变量*/
	pthread_cond_signal(&b->notfull); 
	pthread_mutex_unlock(&b->lock); 
	return data; 
} 
void initmsgqueue(void) {
	sem_init(&sem_offerwait,0,0);
	sem_init(&sem_answerwait,0,0);
    sem_init(&sem_videostreamwait,0,0);
	sem_init(&sem_audiostreamwait,0,0);
	sem_init(&sem_playvideostreamwait,0,0);
	sem_init(&sem_playaudiostreamwait,0,0);
	init(&pmsgbuffer);//信令消息
	init(&pvideo_streambuffer);//视频流
	init(&paudio_streambuffer);//音频流
	init(&pplayvideo_streambuffer);//视频流
	init(&pplayaudio_streambuffer);//音频流
}
void closemsgqueue(void) {
    sem_destroy(&sem_answerwait);
    sem_destroy(&sem_offerwait);
	sem_destroy(&sem_videostreamwait);
	sem_destroy(&sem_audiostreamwait);
	sem_destroy(&sem_playvideostreamwait);
	sem_destroy(&sem_playaudiostreamwait);	
}
int sendkvsmsg(pMessage pmsg){
	put(&pmsgbuffer,pmsg);
	sem_post(&sem_offerwait);
	printf("sendkvsmsg %s\r\n",pmsg->Mode);
	return 0;
}
int receivekvsmsg(pMessage  *ppmsg){
	sem_wait(&sem_offerwait);
	*ppmsg = get(&pmsgbuffer);
	printf("receivekvsmsg %s\r\n",(*ppmsg)->Mode);
	return 0;
}
 
int sendkvsstream(ptstreamInfo pstream){
	switch(pstream->streamtype){
		case STREAM_VIDEO:
		    // sem_wait(&sem_videostreamwait);
			putstream(&pvideo_streambuffer,pstream);
			sem_post(&sem_videostreamwait);
			// printf("sendkvsstream %d-> %d\r\n",pstream->streamtype,pstream->size);
		break;
        case STREAM_AUDIO:
			putstream(&paudio_streambuffer,pstream);
			sem_post(&sem_audiostreamwait);
			// printf("sendkvsstream %d-> %d\r\n",pstream->streamtype,pstream->size);
		break;
		default:
		break;
		}       
	return 0;
}
int receivekvsvideostream(ptstreamInfo  *pstream){
	sem_wait(&sem_videostreamwait);
	*pstream = getstream(&pvideo_streambuffer);
	// sem_post(&sem_videostreamwait);
	// printf("receivekvsvideostream %d->%d\r\n",(*pstream)->streamtype,(*pstream)->size);
	return 0;
}
int receivekvsaudiostream(ptstreamInfo  *pstream){
	sem_wait(&sem_audiostreamwait);
	*pstream = getstream(&paudio_streambuffer);
	// printf("receivekvsaudiostream %d->%d\r\n",(*pstream)->streamtype,(*pstream)->size);
	return 0;
}
//play
int sendkvsplaystream(ptstreamInfo pstream){
	switch(pstream->streamtype){
		case STREAM_VIDEO:
		    // sem_wait(&sem_videostreamwait);
			putstream(&pplayvideo_streambuffer,pstream);
			sem_post(&sem_playvideostreamwait);
			// printf("sendkvsstream %d-> %d\r\n",pstream->streamtype,pstream->size);
		break;
        case STREAM_AUDIO:
			putstream(&pplayaudio_streambuffer,pstream);
			sem_post(&sem_playaudiostreamwait);
			// printf("sendkvsstream %d-> %d\r\n",pstream->streamtype,pstream->size);
		break;
		default:
		break;
		}       
	return 0;
}
int receivekvspalyvideostream(ptstreamInfo  *pstream){
	sem_wait(&sem_playvideostreamwait);
	*pstream = getstream(&pplayvideo_streambuffer);
	// sem_post(&sem_videostreamwait);
	// printf("receivekvsvideostream %d->%d\r\n",(*pstream)->streamtype,(*pstream)->size);
	return 0;
}
int receivekvsplayaudiostream(ptstreamInfo  *pstream){
	sem_wait(&sem_playaudiostreamwait);
	*pstream = getstream(&pplayaudio_streambuffer);
	// printf("receivekvsaudiostream %d->%d\r\n",(*pstream)->streamtype,(*pstream)->size);
	return 0;
}