Producer consumer model of concurrent model

producer - Consumer model

Producer consumer pattern is a classic multi thread design pattern .

summary ：

1. The producer thread submits the task to the memory buffer , The consumer thread gets the task from the memory buffer and executes .
2. Through memory buffer , Avoid direct communication between producers and consumers , Thus decoupling producers and consumers .
3. Through memory buffer , Allow performance differences between producers and consumers .

stay JDK Thread pool provided in (ThreadPoolExecutor) It is a typical producer consumer model （ The task is thread ）, The implementation of memory buffer uses BlockingQueue Blocking queues .

producer - Consumer model （ No lock implementation ）

stay ThreadPoolExecutor Used in BlockingQueue Block the queue to make the memory buffer , However, due to the use of locks and blocking waiting to achieve synchronization between threads , So the new energy is not high .
and LMAX The company has developed a framework of high-performance producer consumer mode without lock , be called Disruptor.

Example ：（ Producers generate data , The consumer calculates the square of the data ）

Introduce dependencies ：

<dependency>
    <groupId>com.lmax</groupId>
    <artifactId>disruptor</artifactId>
    <version>3.4.4</version>
</dependency>

Data entity ：

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

Data factory ：

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

producer ：

import com.lmax.disruptor.RingBuffer;
import java.nio.ByteBuffer;
public class Producer {
    private final RingBuffer<PCData> ringBuffer;

    public Producer(RingBuffer<PCData> ringBuffer){
        this.ringBuffer = ringBuffer;
    }
    public void pushData(ByteBuffer bb){
        //  Get the next sequence on the ring 
        long sequence = ringBuffer.next();
        PCData data = ringBuffer.get(sequence);
        //  Set up the data 
        data.setValue(bb.getLong(0));
        //  Release sequence 
        ringBuffer.publish(sequence);
    }
}

consumer ：

import com.lmax.disruptor.WorkHandler;
public class Consumer implements WorkHandler<PCData> {
    @Override
    public void onEvent(PCData pcData) throws Exception {
        //  Print the square value 
        System.out.println(Thread.currentThread().getName() +
                " -- value="+pcData.getValue() +
                " --  square ="+Math.pow(pcData.getValue(),2));
    }
}

client ：

public class Main {
    public static void main(String[] args) throws InterruptedException {
        //  The size needs to be 2 The power of 
        int bufferSize = 1024;
        Disruptor<PCData> disruptor = new Disruptor<>(
                new PCDataFactory(),
                bufferSize,
                Executors.defaultThreadFactory(),
                ProducerType.MULTI,
                //  Choose the right strategy , Improve consumer response time 
                new BlockingWaitStrategy() //  Blocking waiting strategy 
                // new SleepingWaitStrategy() //  Sleep waiting strategy 
                // new YieldingWaitStrategy() //  Humble waiting strategy 
                // new BusySpinWaitStrategy() //  Busy spin wait strategy , Dead cycle 
        );
        // 4 Consumers 
        disruptor.handleEventsWithWorkerPool(
                new Consumer(),
                new Consumer(),
                new Consumer(),
                new Consumer()
        );
        disruptor.start();

        //  Generate the data 
        RingBuffer<PCData> ringBuffer = disruptor.getRingBuffer();
        long size = 1000L;
        // 2 A producer 
        new Thread(()->{
            Producer producer = new Producer(ringBuffer);
            ByteBuffer bb = ByteBuffer.allocate(8);
            for(long i = 0L;i<size;i++){
                bb.putLong(0,i);
                producer.pushData(bb);
                System.out.println(Thread.currentThread().getName() + " -  Generate data :"+i);
            }
        }).start();
        new Thread(()->{
            Producer producer = new Producer(ringBuffer);
            ByteBuffer bb = ByteBuffer.allocate(8);
            for(long i = size;i<2*size;i++){
                bb.putLong(0,i);
                producer.pushData(bb);
                System.out.println(Thread.currentThread().getName() + " -  Generate data :"+i);
            }
        }).start();
    }
}

summary ：

1. Choose the right strategy , Improve consumer response time

new BlockingWaitStrategy() //  Blocking waiting strategy , province CPU
new SleepingWaitStrategy() //  Sleep waiting strategy , Moderate delay , Sleep after spin waiting fails , Don't take up too much CPU
new YieldingWaitStrategy() //  Humble waiting strategy , Low latency ,CPU The physical core is larger than the number of threads 
new BusySpinWaitStrategy() //  Busy spin wait strategy , Dead cycle , Eat all CPU resources 

2. Disruptor Yes Sequence Use the method of aligned filling to solve CPU Cache pseudo sharing problem .

CPU Cache pseudo share

Look at the picture below , To know CPU Cache pseudo share The problem of

You can align the stored data to the cache row using padding （64 byte ） size , Only one data is stored in each cache line .

The following code snippet is Disruptor in Sequence inherited RhsPadding class , It is filled with 7 individual long Type value （ One long type 64 Bitwise is 8 byte , repair 7 One plus one of my own 8 individual , total 64 byte , It just takes up a cache line size ）

class RhsPadding extends Value {
    protected long p9;
    protected long p10;
    protected long p11;
    protected long p12;
    protected long p13;
    protected long p14;
    protected long p15;

    RhsPadding() {
    }
}

Reference material

• Books Ge Yiming * 《Java High concurrency programming 》