MPSC lock free queue

[c实现的队列](http://www.1024cores.net/home/lock-free-algorithms/queues/non-intrusive-mpsc-node-based-queue)

下面是akka实现的一个MPSC队列。

PS: 代码中注释对链头链尾判定的标准是添加的元素所在的位置为链尾,这和代码中的命名相冲突了

PPS: single customer 就不太需要考虑消费者的同时取的竞争状态

/**
 * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
 */

package akka.dispatch;

import akka.util.Unsafe;

import java.util.concurrent.atomic.AtomicReference;

/**
 * Lock-free MPSC linked queue implementation based on Dmitriy Vyukov‘s non-intrusive MPSC queue:
 * http://www.1024cores.net/home/lock-free-algorithms/queues/non-intrusive-mpsc-node-based-queue
 */
@SuppressWarnings("serial")
public abstract class AbstractNodeQueue<T> extends AtomicReference<AbstractNodeQueue.Node<T>> {
    // Extends AtomicReference for the "head" slot (which is the one that is appended to) since Unsafe does not expose XCHG operation intrinsically
    //AtomicReference的value对应队列的链尾Node
    @SuppressWarnings("unused")
    private volatile Node<T> _tailDoNotCallMeDirectly;

    protected AbstractNodeQueue() {
       final Node<T> n = new Node<T>();
       _tailDoNotCallMeDirectly = n;//初始化根节点,value=null,链头!
       set(n);//初始化链尾部Node
    }

    /*
     * !!! There is a copy of this code in pollNode() !!!
     */
    @SuppressWarnings("unchecked")
    protected final Node<T> peekNode() {//链头
        for(;;) {
          final Node<T> tail = ((Node<T>)Unsafe.instance.getObjectVolatile(this, tailOffset));
          final Node<T> next = tail.next();
          if (next != null || get() == tail)//next!=null表明它不是根节点且在链表中;get()==tail 表明整个链表中还没有Node,只能返回根节点了
            return next;
        }
    }

    public final T peek() {
        final Node<T> n = peekNode();
        return (n != null) ? n.value : null;
    }

    public final void add(final T value) {
        final Node<T> n = new Node<T>(value);
        getAndSet(n).setNext(n);//蛮经典的一句,将AtomicReference的value设置为最新的Node,并将n链接到链表尾部上去
    }
    
    public final void addNode(final Node<T> n) {
        n.setNext(null);
        getAndSet(n).setNext(n);
    }

    public final boolean isEmpty() {
        return peek() == null;
    }

    public final int count() {//count 是不准确的
        int count = 0;
        for(Node<T> n = peekNode();n != null; n = n.next())
          ++count;
        return count;
    }

    /*
     * !!! There is a copy of this code in pollNode() !!!
     */
    public final T poll() {
        final Node<T> next = peekNode();
        if (next == null) return null;
        else {
            final T ret = next.value;
            next.value = null;
            Unsafe.instance.putOrderedObject(this, tailOffset, next);//将链头替换掉
            return ret;
        }
    }
    
    @SuppressWarnings("unchecked")
    public final Node<T> pollNode() {
      Node<T> tail;
      Node<T> next;
      for(;;) {
        tail = ((Node<T>)Unsafe.instance.getObjectVolatile(this, tailOffset));
        next = tail.next();
        if (next != null || get() == tail)
          break;
      }
      if (next == null) return null;
      else {
        tail.value = next.value;
        next.value = null;
        Unsafe.instance.putOrderedObject(this, tailOffset, next);
        return tail;
      }
    }

    private final static long tailOffset;

    static {
        try {
          tailOffset = Unsafe.instance.objectFieldOffset(AbstractNodeQueue.class.getDeclaredField("_tailDoNotCallMeDirectly"));
        } catch(Throwable t){
            throw new ExceptionInInitializerError(t);
        }
    }

    public static class Node<T> {
        public T value;
        @SuppressWarnings("unused")
        private volatile Node<T> _nextDoNotCallMeDirectly;//下一个节点。next,直接用Unsafe来进行操作

        public Node() {
            this(null);
        }

        public Node(final T value) {
            this.value = value;
        }

        @SuppressWarnings("unchecked")
        public final Node<T> next() {
            return (Node<T>)Unsafe.instance.getObjectVolatile(this, nextOffset);
        }

        protected final void setNext(final Node<T> newNext) {
          Unsafe.instance.putOrderedObject(this, nextOffset, newNext);
        }
        
        private final static long nextOffset;
        
        static {
            try {
                nextOffset = Unsafe.instance.objectFieldOffset(Node.class.getDeclaredField("_nextDoNotCallMeDirectly"));
            } catch(Throwable t){
                throw new ExceptionInInitializerError(t);
            } 
        }
    } 
}
时间: 2024-10-14 00:49:29

MPSC lock free queue的相关文章

练习--python中的Queue与多进程(multiprocessing)

按官方说法: This module is OBSOLETE and is only provided on PyPI to support old projects that still use it. Please DO NOT USE IT FOR NEW PROJECTS! Use modern alternatives like the multiprocessing module in the standard library or even an asynchroneous app

使用C#的泛型队列Queue实现生产消费模式

本篇体验使用C#的泛型队列Queue<T>实现生产消费模式. 如果把生产消费想像成自动流水生产线的话,生产就是流水线的物料,消费就是某种设备对物料进行加工的行为,流水线就是队列. 现在,要写一个体现生产消费模式的泛型帮助类,比如叫ProducerConsumer<T>. 该类肯定会维护一个有关生产.物料的Queue<T>类型的字段,还存在一个有关消费.Action<T>类型的字段. 在ProducerConsumer类的构造函数中,为Action<T&

python多线程--优先级队列(Queue)

Python的Queue模块中提供了同步的.线程安全的队列类,包括FIFO(先入先出)队列Queue,LIFO(后入先出)队列LifoQueue,和优先级队列PriorityQueue.这些队列都实现了锁原语,能够在多线程中直接使用.可以使用队列来实现线程间的同步. Queue模块中的常用方法: Queue.qsize() 返回队列的大小 Queue.empty() 如果队列为空,返回True,反之False Queue.full() 如果队列满了,返回True,反之False Queue.fu

Lock+Condition实现机制

前言:大部分多线程同步场景,在功能和性能层面,synchronized可以满足,少部分场景Lock可以满足,dubbo的源码也符合这个比例,需要使用到Condition的场景极少,整个dubbo源码中只在启动函数中,服务关闭这一处使用到了Lock+Condition机制. 1.Lock+Condition用法 生产者,消费者模式在面试coding中出场率很高,可以用synchronized+wait+ notify来实现,也可以使用Lock+Condition实现.直接上代码 public cl

pipe和queue.py

"""multiprocessing提供了threading包中没有的IPC(进程间通信),效率上更高.应优先考虑Pipe和Queue,避免使用Lock/Event/Semaphore/Condition等同步方式(因为它们占据的不是用户进程的资源).multiprocessing包中有Pipe类和Queue类来分别支持这两种IPC机制.Pipe和Queue可以用来传送常见的对象.①Pip可以是单向(half-duplex),也可以是双向(duplex).我们通过mutipr

转载:用python爬虫抓站的一些技巧总结

原文链接:http://www.pythonclub.org/python-network-application/observer-spider 原文的名称虽然用了<用python爬虫抓站的一些技巧总结>但是,这些技巧不仅仅只有使用python的开发可以借鉴,我看到这篇文章的时候也在回忆自己做爬虫的过程中也用了这些方法,只是当时没有系统的总结而已,谨以此文为鉴,为以前的爬虫程序做一个总结. 转载原文如下: 学用python也有3个多月了,用得最多的还是各类爬虫脚本:写过抓代理本机验证的脚本,

Python爬虫进阶五之多线程的用法

前言 我们之前写的爬虫都是单个线程的?这怎么够?一旦一个地方卡到不动了,那不就永远等待下去了?为此我们可以使用多线程或者多进程来处理. 首先声明一点! 多线程和多进程是不一样的!一个是 thread 库,一个是 multiprocessing 库.而多线程 thread 在 Python 里面被称作鸡肋的存在!而没错!本节介绍的是就是这个库 thread. 不建议你用这个,不过还是介绍下了,如果想看可以看看下面,不想浪费时间直接看 multiprocessing 多进程 鸡肋点 名言: "Pyt

PatentTips - Fair scalable reader-writer mutual exclusion

BACKGROUND The present invention relates generally to multithreaded programming and, more specifically, to mutual exclusion of readers and writers in a multithreaded programming environment. Mutual exclusion is a programming technique that ensures th

Python 学习——多线程

使用Threading模块创建线程: threading模块有Thread类实现threading.Thread类提供的方法如下: run():线程的入口点 start():调用run方法启动线程 join(time):等待线程结束 isAlive():检查一个线程是否仍旧在执行 getName():返回线程的名字 setName():设置一个线程的名字 要使用threading模块实现一个新线程,你得先如下做: 定义Thread类的一个子类. 重写__init__(self,[,args])方