高性能编程：无锁队列----MsgQueue代码实践

概述

代码结构

1. 头文件解析 (msgqueue.h)

2. 实现文件解析 (msgqueue.c)

核心功能解析

2.1 创建队列 (msgqueue_create)

2.2 放入消息 (msgqueue_put)

2.3 获取消息 (msgqueue_get)

2.4 交换队列 (__msgqueue_swap)

2.5 阻塞与非阻塞模式

2.6 销毁队列 (msgqueue_destroy)

3. 测试程序解析 (main_msgqueue.cpp)

关键点解析

4. 工作机制详解

4.1 Put队列和Get队列的分离

4.2 队列为空时的阻塞与交换

4.3 生产者与消费者的碰撞处理

4.4 阻塞与非阻塞模式

5. 示例程序运行流程

6. 运行示例

总结

高性能编程：无锁队列概念https://blog.csdn.net/weixin_43925427/article/details/142203825?sharetype=blogdetail&sharerId=142203825&sharerefer=PC&sharesource=weixin_43925427&sharefrom=from_linkMsgQueue在多生产者多消费者（MPMC）场景下的实现原理及其工作机制。

概述

msgqueue是一个用于多生产者多消费者场景的消息队列实现。它将队列分为put队列和get队列，分别用于生产者和消费者操作。其主要逻辑如下：

Put队列：生产者线程将消息入队到put队列。
Get队列：消费者线程从get队列中出队消息。
交换机制：当get队列为空时，尝试与put队列进行交换，使消费者能够获取新的消息。
阻塞与非阻塞：根据队列的状态和配置，决定是否阻塞消费者或生产者线程。

代码结构

代码分为三个主要部分：

头文件 (msgqueue.h)：定义了消息队列的接口和数据结构。
实现文件 (msgqueue.c)：实现了消息队列的功能。
测试程序 (main_msgqueue.cpp)：演示了如何使用消息队列。

1. 头文件解析 (`msgqueue.h`)

#ifndef _MSGQUEUE_H_
#define _MSGQUEUE_H_

#include <stddef.h>

typedef struct __msgqueue msgqueue_t;

#ifdef __cplusplus
extern "C"
{
#endif

/* 消息队列的接口函数 */
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);

#ifdef __cplusplus
}
#endif

#endif

msgqueue_create：创建一个消息队列，maxlen表示队列的最大长度，linkoff表示消息结构中用于链接下一个消息的偏移量。
msgqueue_put：生产者将消息放入队列。
msgqueue_get：消费者从队列中获取消息。
msgqueue_set_nonblock / msgqueue_set_block：设置队列为非阻塞或阻塞模式。
msgqueue_destroy：销毁消息队列，释放资源。

2. 实现文件解析 (`msgqueue.c`)

#include <errno.h>
#include <stdlib.h>
#include <pthread.h>
#include "msgqueue.h"

/* 消息队列的内部结构 */
struct __msgqueue
{
    size_t msg_max;          // 队列的最大长度
    size_t msg_cnt;          // 当前消息数量
    int linkoff;             // 链接偏移量
    int nonblock;            // 是否为非阻塞模式
    void *head1;             // get队列的第一个头节点
    void *head2;             // put队列的第一个头节点
    void **get_head;         // 指向get队列头指针的指针
    void **put_head;         // 指向put队列头指针的指针
    void **put_tail;         // 指向put队列尾指针的指针
    pthread_mutex_t get_mutex; // get队列的互斥锁
    pthread_mutex_t put_mutex; // put队列的互斥锁
    pthread_cond_t get_cond;   // get队列的条件变量
    pthread_cond_t put_cond;   // put队列的条件变量
};

核心功能解析

2.1 创建队列 (`msgqueue_create`)

msgqueue_t *msgqueue_create(size_t maxlen, int linkoff)
{
    msgqueue_t *queue = (msgqueue_t *)malloc(sizeof(msgqueue_t));
    int ret;

    if (!queue)
        return NULL;

    ret = pthread_mutex_init(&queue->get_mutex, NULL);
    if (ret == 0)
    {
        ret = pthread_mutex_init(&queue->put_mutex, NULL);
        if (ret == 0)
        {
            ret = pthread_cond_init(&queue->get_cond, NULL);
            if (ret == 0)
            {
                ret = pthread_cond_init(&queue->put_cond, NULL);
                if (ret == 0)
                {
                    queue->msg_max = maxlen;
                    queue->linkoff = linkoff;
                    queue->head1 = NULL;
                    queue->head2 = NULL;
                    queue->get_head = &queue->head1;
                    queue->put_head = &queue->head2;
                    queue->put_tail = &queue->head2;
                    queue->msg_cnt = 0;
                    queue->nonblock = 0;
                    return queue;
                }

                pthread_cond_destroy(&queue->get_cond);
            }

            pthread_mutex_destroy(&queue->put_mutex);
        }

        pthread_mutex_destroy(&queue->get_mutex);
    }

    errno = ret;
    free(queue);
    return NULL;
}

初始化互斥锁和条件变量：为get和put操作分别初始化互斥锁和条件变量。
初始化队列头指针：get_head指向head1，put_head和put_tail指向head2，实现put和get队列的分离。

2.2 放入消息 (`msgqueue_put`)

void msgqueue_put(void *msg, msgqueue_t *queue)
{
    void **link = (void **)((char *)msg + queue->linkoff);

    *link = NULL;
    pthread_mutex_lock(&queue->put_mutex);
    while (queue->msg_cnt > queue->msg_max - 1 && !queue->nonblock)
        pthread_cond_wait(&queue->put_cond, &queue->put_mutex);

    *queue->put_tail = link;
    queue->put_tail = link;
    queue->msg_cnt++;
    pthread_mutex_unlock(&queue->put_mutex);
    pthread_cond_signal(&queue->get_cond);
}

计算消息的链接指针：根据linkoff偏移量获取消息结构中用于链接下一个消息的指针。
设置链接指针为空：表示当前消息是链表的末尾。
锁定put队列：确保多个生产者线程安全地操作put队列。
阻塞条件：如果当前消息数量超过msg_max - 1且不是非阻塞模式，生产者线程将被阻塞，直到有空间可用。
入队操作：
- 将当前消息链接到put队列的尾部。
- 更新put队列的尾指针。
- 增加消息计数。
解锁put队列：释放锁，允许其他生产者继续入队。
通知消费者：通过条件变量get_cond唤醒可能被阻塞的消费者线程。

2.3 获取消息 (`msgqueue_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;

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

锁定get队列：确保多个消费者线程安全地操作get队列。
判断get队列是否有消息：
- 如果*get_head不为空，直接获取消息。
- 否则，调用__msgqueue_swap尝试将put队列和get队列交换。
获取消息：
- 计算消息的实际地址（通过减去linkoff偏移量）。
- 更新get队列的头指针，指向下一个消息。
解锁get队列：释放锁，允许其他消费者继续出队。
返回消息：如果队列为空，返回NULL。

2.4 交换队列 (`__msgqueue_swap`)

static size_t __msgqueue_swap(msgqueue_t *queue)
{
    void **get_head = queue->get_head;
    size_t cnt;

    queue->get_head = queue->put_head;
    pthread_mutex_lock(&queue->put_mutex);
    while (queue->msg_cnt == 0 && !queue->nonblock)
        pthread_cond_wait(&queue->get_cond, &queue->put_mutex);

    cnt = queue->msg_cnt;
    if (cnt > queue->msg_max - 1)
        pthread_cond_broadcast(&queue->put_cond);

    queue->put_head = get_head;
    queue->put_tail = get_head;
    queue->msg_cnt = 0;
    pthread_mutex_unlock(&queue->put_mutex);
    return cnt;
}

交换get_head与put_head：将put队列的头指针赋值给get队列，使消费者能够获取新的消息。
锁定put队列：在交换过程中，确保生产者不会同时操作put队列。
阻塞条件：如果put队列为空且不是非阻塞模式，消费者线程将被阻塞，直到有新的消息入队。
获取消息计数：记录当前put队列中的消息数量。
条件广播：如果消息数量超过msg_max - 1，通过put_cond唤醒被阻塞的生产者线程。
重置put队列：
- 将put队列的头尾指针指向之前的get_head。
- 重置消息计数。
解锁put队列：允许生产者线程继续入队。
返回消息计数：用于判断是否有新消息可供获取。

2.5 阻塞与非阻塞模式

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);
}

void msgqueue_set_block(msgqueue_t *queue)
{
    queue->nonblock = 0;
}

msgqueue_set_nonblock：
- 设置队列为非阻塞模式。
- 通过条件变量唤醒所有被阻塞的生产者和消费者线程，防止死锁。
msgqueue_set_block：
- 设置队列为阻塞模式。

2.6 销毁队列 (`msgqueue_destroy`)

void msgqueue_destroy(msgqueue_t *queue)
{
    pthread_cond_destroy(&queue->put_cond);
    pthread_cond_destroy(&queue->get_cond);
    pthread_mutex_destroy(&queue->put_mutex);
    pthread_mutex_destroy(&queue->get_mutex);
    free(queue);
}

销毁互斥锁和条件变量：释放资源。
释放队列内存：销毁队列对象。

3. 测试程序解析 (`main_msgqueue.cpp`)

#include "msgqueue.h"

#include <cstddef>
#include <thread>
#include <iostream>

// 消息结构体
struct Count {
    Count(int _v) : v(_v), next(nullptr) {}
    int v;
    Count *next;
};

int main() {
    // linkoff: Count结构体中用于链接下一个节点的指针的偏移量
    msgqueue_t* queue = msgqueue_create(1024, sizeof(int));

    // 生产者线程1
    std::thread pd1([&]() {
        msgqueue_put(new Count(100), queue);
        msgqueue_put(new Count(200), queue);
        msgqueue_put(new Count(300), queue);
        msgqueue_put(new Count(400), queue);
    });

    // 生产者线程2
    std::thread pd2([&]() {
        msgqueue_put(new Count(500), queue);
        msgqueue_put(new Count(600), queue);
        msgqueue_put(new Count(700), queue);
        msgqueue_put(new Count(800), queue);
    });

    // 消费者线程1
    std::thread cs1([&]() {
        Count *cnt;
        while((cnt = (Count *)msgqueue_get(queue)) != NULL) {
            std::cout << std::this_thread::get_id() << " : pop " << cnt->v << std::endl;
            delete cnt;
        }
    });

    // 消费者线程2
    std::thread cs2([&]() {
        Count *cnt;
        while((cnt = (Count *)msgqueue_get(queue)) != NULL) {
            std::cout << std::this_thread::get_id() << " : pop " << cnt->v << std::endl;
            delete cnt;
        }
    });

    // 等待所有线程完成
    pd1.join();
    pd2.join();
    cs1.join();
    cs2.join();

    // 销毁队列
    msgqueue_destroy(queue);
    
    return 0;
}