无锁消息队列的设计实现

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

无锁队列的需求分析:

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

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

目录
相关文章
|
2月前
|
消息中间件
无锁消息队列
无锁消息队列
22 0
|
2月前
|
消息中间件 存储 算法
无锁消息队列的实现
无锁消息队列的实现
|
3天前
|
消息中间件 测试技术 RocketMQ
消息队列 MQ产品使用合集之在异步发送消息函数sendMessage()中出现了错误,错误代码为-3,该如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
3天前
|
消息中间件 网络协议 RocketMQ
消息队列 MQ产品使用合集之broker开启proxy,启动之后producer生产消息始终都只到一个broker,该怎么办
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
3天前
|
消息中间件 网络安全 开发工具
消息队列 MQ产品使用合集之使用grpc proxy,生产者心跳并没有发送至Default中,如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
3天前
|
消息中间件 开发工具 RocketMQ
消息队列 MQ产品使用合集之如何关闭客户端的日志记录
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
3天前
|
消息中间件 监控 Oracle
消息队列 MQ产品使用合集之启动Namesrv节点时,遇到报错,该如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
3天前
|
消息中间件 Java RocketMQ
消息队列 MQ产品使用合集之当SpringBoot应用因网络不通而启动失败时,该如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
3天前
|
消息中间件 监控 Java
消息队列 MQ产品使用合集之如何查看推送是否被限制
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
3天前
|
消息中间件 存储 开发工具
消息队列 MQ产品使用合集之C++如何使用Paho MQTT库进行连接、发布和订阅消息
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。

热门文章

最新文章