Example of producer consumer code implemented by the destructor framework without lock

（ Excerpt from 《 actual combat Java High concurrency programming 》）
Disruptor Framework is made up of LMAX An efficient unlocked memory queue developed by the company . It implements a ring queue in a lock free way , It is very suitable for realizing the producer and consumer model , Such as the release of events and news . stay Disruptor in , Ingeniously used the ring queue （RingBuffer） To replace ordinary linear queues , This circular queue is internally implemented as a normal array . For a general queue , It's bound to provide queue synchronization head And the tail tail Two pointers , Used to get out and get in , This undoubtedly increases the complexity of thread collaboration . But if the queue is circular , You only need to provide a current location cursor, Using this pointer, you can enter the queue or exit the queue . Because of the circular queue , The total size of the queue must be specified in advance , Can't dynamically expand . To be able to quickly start from a sequence （sequence） Corresponding to the actual position of the array （ Every time there are elements in the team , The sequence adds 1）,Disruptor It requires that we must set the size of the array to 2 Integer power of . This way sequence&（queueSize-1） You can immediately locate the actual element position index. This is better than the remainder （%） It's much faster .

Another advantage of this fixed size ring queue is that it can achieve complete memory reuse . During the operation of the system , There will be no new space to allocate or old space to recycle . therefore , It can greatly reduce the additional overhead of system allocation space and recycling space .

Code ：
Data class ：

public class PData {
    
    private long value;

    public void setValue(long value) {
    
        this.value = value;
    }
    public long getValue() {
    
        return value;
    }
}

Data production factory ：

public class PFactory implements EventFactory<PData> {
    
    @Override
    public PData newInstance() {
    
        return new PData();
    }
}

consumer ：

public class Consumer implements WorkHandler<PData> {
    

    @Override
    public void onEvent(PData pData) throws Exception {
    
        System.out.println(" Threads ID："+Thread.currentThread().getId()+": Consumption data  "+pData.getValue());
    }
}

Producers and main functions ：

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/** * @author hzy * @date 2022-07-21 */
public class Producer {
    
    private final RingBuffer<PData> ringBuffer;

    public Producer(RingBuffer<PData> ringBuffer) {
    
        this.ringBuffer = ringBuffer;
    }
    public void pushData(ByteBuffer buffer) {
    
        long sequence = ringBuffer.next();
        try{
    
            PData data = ringBuffer.get(sequence);
            data.setValue(buffer.getLong(0));
        }catch(Exception e){
    }
        finally{
    
            ringBuffer.publish(sequence);
        }
    }

    public static void main(String[] args) throws InterruptedException {
    
        Executor excutor = Executors.newCachedThreadPool();
        EventFactory factory =new  PFactory();
        int bufferSize = 1024;
        Disruptor<PData>disruptor = new Disruptor<PData>(factory,
                bufferSize,
                excutor,
                ProducerType.MULTI,
                new BlockingWaitStrategy());
        disruptor.handleEventsWithWorkerPool(new Consumer(),
                new Consumer(),
                new Consumer(),
                new Consumer());
        disruptor.start();
        RingBuffer<PData>ringBuffer = disruptor.getRingBuffer();
        Producer producer = new Producer(ringBuffer);
        ByteBuffer byteBuffer = java.nio.ByteBuffer.allocate(8);
        for(long value = 0L;value < 10L;value++){
    
            byteBuffer.putLong(0,value);
            producer.pushData(byteBuffer);
            Thread.sleep(10);
            System.out.println("add a data"+value);
        }
    }
}

Running results ：