Disruptor:高效的并发组件(框架),可以认为是没有锁的生产者消费者模型,所以效率高,内部使用RingBuffer机制,做到无锁并发,
代码示例
Disruptor服务器
package com.gbcom.frame.disruptor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventTranslatorThreeArg;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
/**
* DisruptorServer,封装了Disruptor服务,提供 start ,stop,注册消费者,提供发布方法等功能
*
*
* 该类 可以直接使用,并且独立于业务。该类描述了 Disruptor的核心
* @author syz
* @date 2014-10-17
* @version v1.0.0
* @see com.gbcom.frame.disruptor.DisruptorServer
*/
public class DisruptorServer {
private static final int BUFFER_SIZE = 4;
private Disruptor<FrameEvent> disruptor;
private EventProcessor processor = new EventProcessor();
/**
* 数据的封装;封装成 FrameEvent
*/
private static final EventTranslatorThreeArg<FrameEvent, String, Integer, byte[]> TRANSLATOR = new EventTranslatorThreeArg<FrameEvent, String, Integer, byte[]>(){
@Override
public void translateTo(FrameEvent event, long seq, String ip,
Integer port, byte[] data) {
event.setData(data);
event.setIp(ip);
event.setPort(port);
}
};
private static class DisruptorHolder{
private static final DisruptorServer INSTANCE = new DisruptorServer();
}
public static DisruptorServer getInstance(){
return DisruptorHolder.INSTANCE;
}
public void start(){
initDisruptor();
}
private void initDisruptor() {
ExecutorService executor = Executors.newCachedThreadPool(); //线程池,构建disruptor需要
disruptor = new Disruptor<FrameEvent>(FrameEvent.ENENT_FACTORY, BUFFER_SIZE, executor, ProducerType.SINGLE, new BlockingWaitStrategy());//构建disruptor
disruptor.handleEventsWith(new FrameEventHandler<FrameEvent>(processor));//注册一个消费者,消费者事先注册好
disruptor.start();//开启并发服务
disruptor.getRingBuffer();//启动轮转,轮转的size 在构建disruptor指定,基于该机制实现的无锁 高并发
}
/**
* 生产者,发布一个消息: (DisruptorServer.onData)
* @param ip
* @param port
* @param data
*/
public void onData(String ip,Integer port,byte[] data){
disruptor.getRingBuffer().publishEvent(TRANSLATOR,ip,port,data);
}
/**
* 生产错误的处理。直接调用,非添加到队列。
* @param t
*/
public void onError(Throwable t){
if(processor !=null){
processor.onError(t);
}
}
}
disruptor 框架的事件
package com.gbcom.frame.disruptor;
import java.util.Arrays;
import com.lmax.disruptor.EventFactory;
/**
* disruptor 框架的事件。POJO,disruptor 需要提供factory,
* @author syz
* @date 2014-10-17
* @version v1.0.0
* @see com.gbcom.frame.disruptor.FrameEvent
*/
public class FrameEvent
{
private byte[] data;
private String ip;
private int port;
public byte[] getData() {
return data;
}
public void setData(byte[] data) {
this.data = data;
}
public String getIp() {
return ip;
}
public void setIp(String ip) {
this.ip = ip;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
/**
* 工厂类
*/
public static final EventFactory<FrameEvent> ENENT_FACTORY = new EventFactory<FrameEvent> (){
@Override
public FrameEvent newInstance() {
// TODO Auto-generated method stub
return new FrameEvent();
}
};
@Override
public String toString() {
return "FrameEvent [data=" + Arrays.toString(data) + ", ip=" + ip
+ ", port=" + port + "]";
}
}
FrameEventHandler实现EventHandler接口,消费者处理
package com.gbcom.frame.disruptor;
import com.lmax.disruptor.EventHandler;
/**
* 消费者处理,,实现EventHandler接口
* @author SYZ
* @date 2016-11-3 下午02:08:32
* @version 1.0.0
* @see com.gbcom.frame.disruptor.FrameEventHandler
*/
public class FrameEventHandler<FrameEvent> implements EventHandler<FrameEvent>{
private EventProcessor processor ;
protected FrameEventHandler(EventProcessor processor) {
super();
this.processor = processor;
}
@Override
public void onEvent(FrameEvent event, long seq, boolean endOfBatch)
throws Exception {
processor.onReceive((com.gbcom.frame.disruptor.FrameEvent) event);
}
}
消费者:真正的消息处理
package com.gbcom.frame.disruptor;
import java.io.UnsupportedEncodingException;
import org.apache.commons.io.Charsets;
import com.gbcom.frame.disruptor.snetty.UdpHandler;
/**
* 消费者:真正的消息处理
* @author syz
* @date 2014-10-17
* @version v1.0.0
* @see com.gbcom.frame.disruptor.EventProcessor
*/
public class EventProcessor implements UdpHandler{
public void onReceive(FrameEvent event){
System.out.println("receive is success :"+event);
try {
NatTabManager.getInstance().put(new String(event.getData(),"utf-8"), event);
} catch (UnsupportedEncodingException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
event = NatTabManager.getInstance().getNat(new String(event.getData(),Charsets.UTF_8));
//System.out.println(HttpClientUtil.get("http://"+event.getIp()+":"+event.getPort()));
}
public void onError(Throwable t ){
System.out.println("receive is false :" + t);
}
public static void main(String args[]){
System.out.println(HttpClientUtil.get("http://127.0.0.1:80"));
}
@Override
public void onReceive(String ip, int port, byte[] data) {
}
}
NET功能,,跟Disruptor框架无关,上层应用
package com.gbcom.frame.disruptor;
import java.util.HashMap;
import java.util.Map;
/**
* NET功能,,跟Disruptor框架无关,
* @author SYZ
* @date 2016-11-3 下午02:09:14
* @version 1.0.0
* @see com.gbcom.frame.disruptor.NatTabManager
*/
public class NatTabManager {
private Map<String,FrameEvent> natTab = new HashMap<String,FrameEvent>();
private static NatTabManager intance = new NatTabManager();
public static NatTabManager getInstance(){
return intance;
}
public void put(String key,FrameEvent value){
natTab.put(key, value);
}
public Map getNatTab(){
return natTab;
}
public FrameEvent getNat(String key){
return natTab.get(key);
}
}
客户端
package com.gbcom.frame.disruptor;
/**
*
*
* Disruptor:高效的并发组件(框架),可以认为是没有锁的生产者消费者模型,所以效率高
*
* 该例子完成Disruptor功能,且具有一定的使用价值,Disruptor独立于业务,,,DisruptorServer
*
* 高效的生产者消费者模型:DisruptorServer
*
* @author SunYanzheng
* @date 2014-10-17
* @version v1.0.0
* @see com.gbcom.frame.disruptor.DisClient
*/
public class DisClient {
/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
DisruptorServer.getInstance().start(); //初始化服务
DisruptorServer.getInstance().onData("10.1.1.1", 162, "hello".getBytes());//调用onData 发布数据
}
}
完整代码下载参考附件
分享到:
相关推荐
Disruptor它是一个开源的并发框架,并获得2011 Duke’s 程序框架创新奖,能够在无锁的情况下实现网 络的Queue并发操作。本文是Disruptor官网中发布的文章的译文(现在被移到了GitHub)。
在官方原有基础上为降低cpu占用修改版本,性能方面没有任何影响
springboot整合Disruptor并发编程框架,适合高并发场景需求的开发者
Java并发框架Disruptor,里面采取环形缓存结构,速度更快,适用于生产者消费者模式
Disruptor 是一个 Java 的并发编程框架,大大的简化了并发程序开发的难度,在性能上也比 Java 本身提供的一些并发包要好。 标签:Disruptor
资源内容:高并发框架Disruptor版本3.3.2,以及框架API使用及测试代码;
Disruptor它是一个开源的并发框架能够在无锁的情况下实现网络的Queue并发操作。同时,Disruptor是一个高性能的异步处理框架,或者可以认为是最快的消息框架(轻量的JMS),也可以认为是一个观察者模式的实现,或者...
Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip
LMAX是一种新型零售金融交 易平台,它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上,其核心是一个业务逻辑处理 器,它能够在一个线程里每秒处理6百万订单。...业务逻辑处理器的核心是Disruptor。
Disruptor它是一个开源的并发框架,并获得2011 Duke’s 程序框架创新奖,能够在无锁的情况下实现网络的Queue并发操作。 Disruptor是一个高性能的异步处理框架,或者可以认为是最快的消息框架(轻量的JMS),也可以...
xd-toolkit 工具集(工作过程中常用到的工具类、工具模块) SPMC并发组件(同时也支持MPMC模式) ...LMAX Disruptor并发框架封装 应用场景: a、原框架使用起来比较繁杂,其实很多时候我们只关心生
不错的框架,可以好好研究研究,速度下载,速度下载速度下载速度下载
最快的java并发框架disruptor已经将编译好的jar包集成在里面过阵子写个完整版本的示例再上传,本人其他资源也不错请大家多多下载
disruptor-3.4.2.jar 工具jar包 及 disruptor-3.4.2-sources.jar, Disruptor它是一个开源的并发框架,并获得2011 Duke’s 程序框架创新奖,能够在无锁的情况下实现网络的Queue并发操作,是 log4j2 引用的 jar 包
其实Disruptor与其说是一个框架,不如说是一种设计思路,这个设计思路对于存在“并发、缓冲区、生产者—消费者模型、事务处理”这些元素的程序来说,Disruptor提出了一种大幅提升性能(TPS)的方案。
disruptor-3.4.2.jar, 工具jar包, Disruptor它是一个开源的并发框架,并获得2011 Duke’s 程序框架创新奖,能够在无锁的情况下实现网络的Queue并发操作
Producer:生产者,泛指通过Disruptor发布事件的用户代码(实际业务代码,而并发框架代码)生成Event数据。 Event:事件,从生产者到消费者过程中的数据单元。由用户定义代码。 EventHandler:消费者,代表...
并发框架Disruptor介绍Martin Fowler在自己网站上写了一篇LMAX架构的文章,在文章中他介绍了LMAX是一种新型零售金融交易平台,它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上,其核心是一个业务逻辑...
High performance alternative to bounded queues for exchanging data between concurrent threads Disruptor 框架作者的论文。