[数据结构与算法]搜狗workflow项目研究（八）消息队列

2021SC@SDUSC

目录

一、基本接口

二、初始化

?三、put数据

四、get数据

五、数据结构

一、基本接口

msgqueue_t *msgqueue_create(size_t maxlen, int linkoff);
void msgqueue_put(void *msg, msgqueue_t *queue);
void *msgqueue_get(msgqueue_t *queue);
void msgqueue_set_nonblock(msgqueue_t *queue);
void msgqueue_set_block(msgqueue_t *queue);
void msgqueue_destroy(msgqueue_t *queue);

*msgqueue_create：创建消息队列

msgqueue_put：put数据

*msgqueue_get：get数据

msgqueue_set_nonblock：设置存放数据方式为非堵塞

msgqueue_set_block：设置存放数据方式为堵塞

msgqueue_destroy：销毁消息队列

二、初始化

msgqueue_t *msgqueue_create(size_t maxlen, int linkoff);

在信息列表初始化的时候，需要指定最大消息数量和消息头部偏移量。其中maxlen为最大消息数量，linkoff为消息头部偏移量。

消息头部偏移量主要是为了利用msg头部偏移后的空间，来形成链表。

一个常规的消息队列节点：

struct Node {
	void* your_message;
	Node* next;
};

在这里没有指定节点内容，而是使用了msg头部偏移后的空间，因此在使用的时候，该偏移量的后部需要保留一个指针大小的空间。具体可见下图：

?三、put数据

放数据时分为堵塞和非堵塞两种形式，默认为堵塞。

?堵塞情况：在放数据的时候，如果目前消息队列已超过设定的最大值，那么就等待信息被消费。

非堵塞情况：如果设置了非堵塞，则会释放get_cond和put_cond，通知目前堵塞的线程继续往下执行。此时用户设置的最大值不再生效，会继续向消息队列中放入数据。

void msgqueue_set_nonblock(msgqueue_t *queue)
{
	queue->nonblock = 1;
	pthread_mutex_lock(&queue->put_mutex);
	pthread_cond_signal(&queue->get_cond);
	pthread_cond_broadcast(&queue->put_cond);
	pthread_mutex_unlock(&queue->put_mutex);
}

四、get数据

如果当前队列中有数据，直接读取队列中下一个数据。如果当前队列中没有数据，返回空，设置错误码。

get数据时，返回数据的头部。

void *msgqueue_get(msgqueue_t *queue)
{
	void *msg;

	pthread_mutex_lock(&queue->get_mutex);
	if (*queue->get_head || __msgqueue_swap(queue) > 0)
	{
		// 获取消息头部
        msg = (char *)*queue->get_head - queue->linkoff;
        // 将头部移动到下一个元素上
		*queue->get_head = *(void **)*queue->get_head;
	}
	else
	{
		msg = NULL;
		errno = ENOENT;
	}

	pthread_mutex_unlock(&queue->get_mutex);
	return msg;
}

五、数据结构

struct __msgqueue {
	size_t msg_max;
	size_t msg_cnt;

	// 消息的偏移量
	int linkoff;
	// 是否为堵塞
	int nonblock;

	// 消息队列的两个头部
	void *head1;
	void *head2;

	// get 队列
	void **get_head;

	// put 队列头部和尾部
	void **put_head;
	void **put_tail;

	// 锁
	pthread_mutex_t get_mutex;
	pthread_mutex_t put_mutex;
	pthread_cond_t get_cond;
	pthread_cond_t put_cond;
};