一、Disruptor是啥？

一个无锁的高并发框架，环形的数据结构(直接覆盖（不用清除）旧的数据、大小必须是2的幂次方)，降低GC频率
实现了基于事件的生产者消费者模式（观察者模式）

二、简单的业务场景：

多部门日志文件分析：去指定部门下载日志文件，对日报内容做入库解析

三、业务场景简单图示：

四、简单拆分

    定义产品DataInfo：生产者需要的数据结构

    工厂FtpFactory：数据交互

    消费者ADPT、BDPT、CDPT、DDPT：abcd消费者

    生产者DownloadFile：入参伪造

    构建业务入口：

五、简单实现

    1.添加maven项目依赖

<dependency><groupId>com.lmax</groupId><artifactId>disruptor</artifactId><version>3.2.1</version>
</dependency>
<dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId>
</dependency>

    2.定义产品

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;import java.util.List;
import java.util.Map;/*** @Description* @ClassName DataInfo* @Author Morik* @date 2020.09.28 11:24*/
@Slf4j
@Setter
@Getter
public class DataInfo {Map<String, List<FtpInfo>> datainfo;class FtpInfo {String url;String uri;String name;int delay;String dartment;}
}

3.创建工厂

import com.lmax.disruptor.EventFactory;/*** @Description* @ClassName FtpFactory* @Author Morik* @date 2020.09.28 11:31*/
public class FtpFactory implements EventFactory<DataInfo> {@Overridepublic DataInfo newInstance() {return new DataInfo();}
}

4.定义消费者

import com.lmax.disruptor.EventHandler;/*** @Description* @ClassName ADPT* @Author Morik* @date 2020.09.28 11:30*/
public class ADPT implements EventHandler<DataInfo> {@Overridepublic void onEvent(DataInfo dataInfo, long l, boolean b) throws Exception {}
}

5.定义生成者

import com.lmax.disruptor.RingBuffer;/*** @Description* @ClassName DownloadFile* @Author Morik* @date 2020.09.28 11:45*/
public class DownloadFile {RingBuffer<DataInfo> ringBuffer;public DownloadFile(RingBuffer<DataInfo> ringBuffer) {this.ringBuffer = ringBuffer;}public void downloading() {long sequence = ringBuffer.next();try {//取出空队列DataInfo df = ringBuffer.get(sequence);//获取时间队列传递数据df.setDatainfo(null);} finally {//发布事件ringBuffer.publish(sequence);}}
}

6.构建业务入口

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;/*** @Description* @ClassName LogPaser* @Author Morik* @date 2020.09.28 12:31*/@Service
@EnableScheduling
public class LogPaser {@Scheduled(fixedRate = 30000)  //30s执行一次public void downlogs() {ExecutorService executor = Executors.newCachedThreadPool();EventFactory<DataInfo> eventFactory = new FtpFactory();int ringBufferSize = 2;Disruptor<DataInfo> disruptor =new Disruptor(eventFactory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());disruptor.handleEventsWith(new ADPT(), new ADPT(), new ADPT(), new ADPT(), new ADPT(), new ADPT());//多个银行在这里添加disruptor.start();RingBuffer<DataInfo> ringBuffer = disruptor.getRingBuffer();DownloadFile ftpDownLoding = new DownloadFile(ringBuffer);ftpDownLoding.downloading();disruptor.shutdown();executor.shutdown();}
}