企业官网建设流程全解析

无锁环MPMC
目标：
1、实现多生产者、多消费者无锁队列
2、不用互斥锁
3、数组循环复用
4、FIFO
5、不能丢消息、不能重复读、不能读到脏数据

底层载体：
1、固定数组大小 slots[]
2、capacity 必须是 2 的幂次

#include<stdio.h>#include<stdlib.h>#include<pthread.h>#include<stdatomic.h>#include<string.h>/* PPT无锁环关键点： 1. 环形数组缓存，元素定长，这里用uint32_t（4字节，符合PPT 4字节整数倍） 2. 多生产者、多消费者MPMC 3. 不加锁，全部依靠原子CAS 4. 全局共享：prod_head/prod_tail，cons_head/cons_tail；线程本地私有副本做缓存，减少共享变量访问 */typedefuint32_telem_t;// 定长元素：4字节，和PPT要求匹配typedefstruct{atomic_uint seq;// 槽位序列号，无锁同步核心elem_tdata;}slot_t;typedefstruct{slot_t*slots;size_tcapacity;// 必须是2^Nsize_tmask;// capacity‑1，代替%取模// =========全局共享变量（所有线程可见）========atomic_size_tprod_head;// 生产者headatomic_size_tprod_tail;// 生产者tailatomic_size_tcons_head;// 消费者headatomic_size_tcons_tail;// 消费者tail}lockfree_ring_t;// 创建无锁环，capacity必须为2的幂lockfree_ring_t*ring_create(size_tcapacity){lockfree_ring_t*r=malloc(sizeof(lockfree_ring_t));r->capacity=capacity;r->mask=capacity-1;r->slots=calloc(capacity,sizeof(slot_t));atomic_init(&r->prod_head,0);atomic_init(&r->prod_tail,0);atomic_init(&r->cons_head,0);atomic_init(&r->cons_tail,0);for(size_ti=0;i<capacity;i++){atomic_init(&r->slots[i].seq,i);}returnr;}// 销毁voidring_destroy(lockfree_ring_t*r){free(r->slots);free(r);}/** * 多生产者入队 enqueue * 返回0成功；‑1队列满 */intring_enqueue(lockfree_ring_t*r,elem_tval){size_tpos,next;// 自旋CAS争抢prod_head（多生产者竞争）while(1){pos=atomic_load_explicit(&r->prod_head,memory_order_relaxed);next=pos+1;size_tcons_t=atomic_load_explicit(&r->cons_tail,memory_order_acquire);if((next-cons_t)>r->capacity){return-1;// 队列满}// CAS抢占位置if(atomic_compare_exchange_weak_explicit(&r->prod_head,&pos,next,memory_order_relaxed,memory_order_relaxed)){break;}}size_tidx=pos&r->mask;slot_t*s=&r->slots[idx];// 等待槽位空闲（上一轮消费完成）while(atomic_load_explicit(&s->seq,memory_order_acquire)!=pos){;}s->data=val;atomic_store_explicit(&s->seq,pos+1,memory_order_release);// 更新prod_tail，推进发布指针，允许其他线程看到写入while(!atomic_compare_exchange_weak_explicit(&r->prod_tail,&pos,next,memory_order_release,memory_order_relaxed)){pos=atomic_load_explicit(&r->prod_tail,memory_order_relaxed);}return0;}/** * 多消费者出队 dequeue * 返回0成功；‑1队空 */intring_dequeue(lockfree_ring_t*r,elem_t*out){size_tpos,next;while(1){pos=atomic_load_explicit(&r->cons_head,memory_order_relaxed);next=pos+1;size_tprod_t=atomic_load_explicit(&r->prod_tail,memory_order_acquire);if(pos==prod_t){return-1;// 队列为空}if(atomic_compare_exchange_weak_explicit(&r->cons_head,&pos,next,memory_order_relaxed,memory_order_relaxed)){break;}}size_tidx=pos&r->mask;slot_t*s=&r->slots[idx];// 等待生产者写完数据while(atomic_load_explicit(&s->seq,memory_order_acquire)!=pos+1){;}*out=s->data;atomic_store_explicit(&s->seq,pos+r->capacity,memory_order_release);// 更新cons_tailwhile(!atomic_compare_exchange_weak_explicit(&r->cons_tail,&pos,next,memory_order_release,memory_order_relaxed)){pos=atomic_load_explicit(&r->cons_tail,memory_order_relaxed);}return0;}// =====================测试多生产者多消费者====================#defineRING_CAP(1<<4)// 16，2的幂#definePRODUCER_NUM3#defineCONSUMER_NUM2#definePER_PROD_CNT20lockfree_ring_t*g_ring;/* 函数功能：向无锁环里塞数据的生产者线程 *//* 入参：arg 把 tid 数字本身伪装成指针值，没有对应内存*//* 返回值：现成结束返回值，可以给 pthread_join 拿到 */void*producer_thread(void*arg){longtid=(long)arg;for(inti=0;i<PER_PROD_CNT;i++){elem_tv=tid*1000+i;while(ring_enqueue(g_ring,v)!=0){;// 满则自旋}printf("[P%ld] enqueue: %u\n",tid,v);}returnNULL;}void*consumer_thread(void*arg){longtid=(long)arg;elem_tval;inttotal=PER_PROD_CNT*PRODUCER_NUM;staticatomic_int recv_cnt=0;while(atomic_load(&recv_cnt)<total){if(ring_dequeue(g_ring,&val)==0){atomic_fetch_add(&recv_cnt,1);printf(" [C%ld] dequeue: %u\n",tid,val);}}returnNULL;}intmain(void){g_ring=ring_create(RING_CAP);pthread_tprods[PRODUCER_NUM];pthread_tcons[CONSUMER_NUM];for(inti=0;i<PRODUCER_NUM;i++){pthread_create(&prods[i],NULL,producer_thread,(void*)(long)i);}for(inti=0;i<CONSUMER_NUM;i++){pthread_create(&cons[i],NULL,consumer_thread,(void*)(long)i);}for(inti=0;i<PRODUCER_NUM;i++)pthread_join(prods[i],NULL);for(inti=0;i<CONSUMER_NUM;i++)pthread_join(cons[i],NULL);ring_destroy(g_ring);printf("====all done====\n");return0;}

给坏代码打补丁

/** * 多生产者入队 enqueue * 返回0成功；‑1队列满 */intring_enqueue(lockfree_ring_t*r,elem_tval){size_tpos,next;// 自旋CAS争抢prod_head（多生产者竞争）while(1){pos=atomic_load_explicit(&r->prod_head,memory_order_relaxed);next=pos+1;size_tcons_t=atomic_load_explicit(&r->cons_tail,memory_order_acquire);if((next-cons_t)>r->capacity){return-1;// 队列满}// CAS抢占位置if(atomic_compare_exchange_weak_explicit(&r->prod_head,&pos,next,memory_order_relaxed,memory_order_relaxed)){break;}}size_tidx=pos&r->mask;slot_t*s=&r->slots[idx];// 等待槽位空闲（上一轮消费完成）while(atomic_load_explicit(&s->seq,memory_order_acquire)!=pos){;}s->data=val;atomic_store_explicit(&s->seq,pos+1,memory_order_release);// 更新prod_tail，推进发布指针，允许其他线程看到写入// 以下 for 循环函数是给注释代码打补丁，注释代码存在问题/*while(!atomic_compare_exchange_weak_explicit( &r->prod_tail, &pos, next, memory_order_release, memory_order_relaxed )){ pos = atomic_load_explicit(&r->prod_tail, memory_order_relaxed); }*/for(;;){size_texpected=atomic_load_explicit(&r->prod_tail,memory_order_relaxed);if(expected!=pos){continue;//还没轮到我，等别的线程推进}size_tdesired=expected+1;if(atomic_compare_exchange_weak_explicit(&r->prod_tail,&expected,desired,memory_order_release,memory_order_relaxed)){break;}// CAS失败，expected已经更新，下一轮重新判断}return0;}