生产者-消费者模式

简述

在生产者 - 消费者模式中，通常有两类线程，一类是生产者线程一类是消费者线程。生产者线程负责提交用户请求，消费者线程则负责处理生产者提交的任务。

​ 最简单粗暴的做法就是生产者每提交一个任务，消费者就立即处理，直到所有任务处理完。但是这样直接通信很容易出现性能上的问题，消费者必须等待它的生产者提交到任务才能执行，就不能达到真正的并行。同时生产者和消费者之间存在依赖关系，在设计上耦合度非常高，这是不可取的。那么最好的做法就是加一个中间层作为通信桥梁。

生产者和消费者之间通过共享内存缓存区进行通信。多个生产者线程将任务提交给共享内存缓存区，消费者线程并不直接与生产者线程通信，而在共享内存缓冲区获取任务，并行地处理。其中内存缓冲区的主要功能是数据再多线程间的共享，同时还可以通过该缓存区，缓解生产者和消费者间的性能差。它是生产者消费者模式的核心组件，既能作为通信的桥梁，又能避免两者直接通信，从而将生产者和消费者进行解耦。生产者不需要消费者的存在，消费者也不需要知道生产者的存在。

​ 由于内存缓冲区的存在，允许生产者和消费者在执行速度上有差异，无论哪一方速度超过另一方，缓冲区都可以得到缓解，确保系统正常运行。

实现

使用 Disruptor 作为Queue

生产者 - 消费者模式 UML 图如下：

Disruptor 是第三方包，需要下载 jar 包导入项目或者使用 Maven 导入

Data 类

package thread.producerConsumer.disruptor;

public class Data {

    private long value;

    public long getValue() {
        return value;
    }

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

数据生产工厂

package thread.producerConsumer.disruptor;

import com.lmax.disruptor.EventFactory;

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

生产者 Producer

package thread.producerConsumer.disruptor;

import com.lmax.disruptor.RingBuffer;
import java.nio.ByteBuffer;

public class Producer {

    private final RingBuffer<Data> ringBuffer;

    public Producer(RingBuffer<Data> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void pushData(ByteBuffer bb) {
      
        long sequence = ringBuffer.next();

        try {
            Data even = ringBuffer.get(sequence);
            even.setValue(bb.getLong(0));
        } finally {
            ringBuffer.publish(sequence);
        }
    }
}

消费者 Consumer

package thread.producerConsumer.disruptor;

import com.lmax.disruptor.WorkHandler;
import java.text.MessageFormat;

public class Consumer implements WorkHandler<Data> {
    @Override
    public void onEvent(Data data) throws Exception {
        long res = data.getValue() + 1;
        System.out.println(MessageFormat.format(
                "Data processing: {0} + 1 = {1}", data.getValue(), res));
    }
}

Main 运行主类

package thread.producerConsumer.disruptor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

import java.nio.ByteBuffer;
import java.util.concurrent.ThreadFactory;

public class Main {
    private static final int NUMS = 10;
    private static final int SUM = 1000000;

    public static void main(String[] args) {

        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        long start = System.currentTimeMillis();
        DataFactory factory = new DataFactory();
        int bufferSize = 1024;
        Disruptor<Data> disruptor =
                new Disruptor<>(
                        factory,
                        bufferSize,
                        new ThreadFactory() {
                            @Override
                            public Thread newThread(Runnable r) {
                                return new Thread(r);
                            }
                        });
        //  构造100个消费者
        Consumer[] consumers = new Consumer[NUMS];
        for (int i = 0; i < NUMS; i++) {
            consumers[i] = new Consumer();
        }
        disruptor.handleEventsWithWorkerPool(consumers);
        disruptor.start();
        RingBuffer<Data> ringBuffer = disruptor.getRingBuffer();
        Producer producer = new Producer(ringBuffer);
        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long l = 0; l < SUM; l++) {
            bb.putLong(0, l);
            producer.pushData(bb);
            System.out.println("Successfully produced data: " + l);
        }
        long end = System.currentTimeMillis();
        disruptor.shutdown();
        System.out.println("Total time: " + (end - start));
    }
}