无锁消息队列的设计实现

简介: 无锁消息队列的设计实现

无锁队列的需求分析:

多线程访问共享队列的数据时,必须确保对共享队列操作的原子性,有以下情况:

1.生产者,例如tcp服务器接收到请求信息,需要将请求信息push进共享队列

2.消费者,例如线程池的工作线程,需要从共享队列中pop/get一个请求

这两种操作都要求对队列进行修改

确保原子性方式

1.对队列的修改操作加锁(系统调用),

可以确保共享队列的线程安全,但是性能较低,并且可能造成死锁

2.使用原子变量(c++)

使用硬件提供的锁机制,性能较高

3.使用c++的对原子变量的原子操作时需要考虑内存顺序模型,确保原子操作之间的顺序性。

可以看出,实现共享队列的原子性,绕不开使用锁机制

分析多线程锁竞争情况:

1.生产者和生产者之间,多个生产者线程向共享队列push数据,发生锁的竞争

2.消费者和消费者之间,多个消费者线程从队列中pop数据,发生锁的竞争

3.生产者和消费者之间,生产者线程push数据,消费者线程pop数据,发生锁的竞争

由于既定的业务处理逻辑,多生产者之间的锁竞争无法避免,多消费者之间的锁竞争也无法避免

但是生产者和消费者之间的锁竞争是可以通过队列的设计来实现的,接下来就要介绍优雅的队列分离代码

优化锁竞争(生产者和消费者)

我们可以观察到,各个线程发生锁竞争的原因是:共享一个队列

那么使用多个队列是不是就可以避免竞争了?

不是,使用一个共享队列的原因是,需要确保所有线程使用的数据都是一致的,否则会出现同一个任务被执行两次的情况,而同一个队列逻辑严格的保证了数据的一致性,同步性。使用多个队列需要确保所有队列的数据实时一致,这是很难做到的,并且平添了复杂性

方案:所有消费者和所有生产者各使用一个队列

这样还能确保两个队列的数据实时一致吗?

逻辑:

1.生产者将用户请求pop进生产者队列

2.消费者从消费者队列取出请求

3.当消费者判断消费者队列为空,而生产者队列不空时,交换两个队列

这里的交换操作,实际上就是将生产者队列中的数据转移到消费者队列中,这样一来,生产者和消费者使用的数据是完全同步

抽象:使用两个队列的策略实际上相当于消费者从共享队列中一次取出多个请求,保存在自己的队列中多次使用,减少了pop操作。是不是特别像cpu的缓存,从主存中一次取出多条指令或数据,加快处理效率

代码实现:

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

可以看出“交换队列”是通过改变生产者队列指针消费者队列指针的指向实现的,很巧妙!我在看代码前以为是先copy再删除呢

注意这个交换队列的操作发生在消费者线程中,此时对生产者队列的修改操作会引起生产者和消费者的锁竞争,当然这是队列分离策略的唯一发生此竞争的位置

最核心的减少锁竞争的逻辑已经介绍完,接下来介绍无锁队列的原子操作

1.put(生产者向生产者队列中添加一个节点)

2.get(消费者从消费者队列中取出一个节点)

先介绍队列结构:

struct __msgqueue
{
  size_t msg_max; 
  size_t msg_cnt;
  int linkoff; 
  int nonblock;
  void *head1; // 消费者队列的头指针
  void *head2; // 生产者队列的头指针
  void **get_head; // 消费者队列的头指针
  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; // 生产者的条件变量
};

1.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); // 条件等待
// 被信号唤醒,执行put操作
  *queue->put_tail = link;
  queue->put_tail = link;
  queue->msg_cnt++;
  pthread_mutex_unlock(&queue->put_mutex); // 解锁
  pthread_cond_signal(&queue->get_cond); // 唤醒因为没有任务而阻塞的消费者线程
}

2.get

void *msgqueue_get(msgqueue_t *queue)
{
  void *msg;
  pthread_mutex_lock(&queue->get_mutex); // 上锁保护临界区(修改队列)
  if (*queue->get_head || __msgqueue_swap(queue) > 0) 
        // 如果消费者队列不空,继续执行,如果消费者队列为空,交换队列,如果交换成功(生产者队列不空)则继续执行,否则会阻塞在swap函数!
  {
    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;
}

总结:无锁队列并不是不使用锁的队列,而是使用原子变量和原子操作以及队列优化的策略来提高多线程对共享数据的并发访问的性能

推荐学习 https://xxetb.xetslk.com/s/p5Ibb

目录
相关文章
|
6月前
|
消息中间件
无锁消息队列
无锁消息队列
36 0
|
6月前
|
消息中间件 存储 算法
无锁消息队列的实现
无锁消息队列的实现
|
4月前
|
消息中间件 C语言 RocketMQ
消息队列 MQ操作报错合集之出现"Connection reset by peer"的错误,该如何处理
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
4月前
|
消息中间件 Java C语言
消息队列 MQ使用问题之在使用C++客户端和GBase的ESQL进行编译时出现core dump,该怎么办
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
18天前
|
消息中间件 存储 Kafka
MQ 消息队列核心原理,12 条最全面总结!
本文总结了消息队列的12个核心原理,涵盖消息顺序性、ACK机制、持久化及高可用性等内容。关注【mikechen的互联网架构】,10年+BAT架构经验倾囊相授。
|
2月前
|
消息中间件
手撸MQ消息队列——循环数组
队列是一种常用的数据结构,类似于栈,但采用先进先出(FIFO)的原则。生活中常见的排队场景就是队列的应用实例。在数据结构中,队列通常用数组实现,包括入队(队尾插入元素)和出队(队头移除元素)两种基本操作。本文介绍了如何用数组实现队列,包括定义数组长度、维护队头和队尾下标(front 和 tail),并通过取模运算解决下标越界问题。此外,还讨论了队列的空与满状态判断,以及并发和等待机制的实现。通过示例代码展示了队列的基本操作及优化方法,确保多线程环境下的正确性和高效性。
37 0
手撸MQ消息队列——循环数组
|
3月前
|
消息中间件 存储 缓存
一个用过消息队列的人,竟不知为何要用 MQ?
一个用过消息队列的人,竟不知为何要用 MQ?
158 1
|
4月前
|
消息中间件 开发工具 RocketMQ
消息队列 MQ使用问题之一直连接master失败,是什么原因
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
4月前
|
消息中间件 Prometheus 监控
消息队列 MQ使用问题之如何将旧集群的store目录迁移到新集群
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
4月前
|
消息中间件 安全 PHP
消息队列 MQ使用问题之如何获取PHP客户端代码
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。