`
assertmyself
  • 浏览: 29717 次
  • 性别: Icon_minigender_1
  • 来自: 南京
文章分类
社区版块
存档分类
最新评论

Disruptor并发框架

阅读更多
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并发框架中文参考文档

    ### Disruptor并发框架知识点详解 #### 一、Disruptor简介及原理 **Disruptor** 是一款高性能、低延迟的并发框架,它通过无锁设计实现了高效的队列操作,从而大大提升了多线程环境下的性能表现。该框架于2011年...

    disruptor 并发框架

    在官方原有基础上为降低cpu占用修改版本,性能方面没有任何影响

    springboot整合Disruptor并发编程框架 #资源达人分享计划#

    SpringBoot整合Disruptor并发编程框架是针对高并发场景下性能优化的一种技术实践。Disruptor是由LMAX公司开发的一款高性能、低延迟的并发工具,它通过消除线程间的锁竞争,大大提升了多线程环境下的处理速度。...

    Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip

    Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip Netty整合并发编程框架Disruptor实战百万长链接服务构建源码.zip

    disruptor框架案例.rar

    Disruptor它是一个开源的并发框架能够在无锁的情况下实现网络的Queue并发操作。同时,Disruptor是一个高性能的异步处理框架,或者可以认为是最快的消息框架(轻量的JMS),也可以认为是一个观察者模式的实现,或者...

    基于Spring Boot和LMAX Disruptor的高性能并发框架.zip

    基于Spring Boot和LMAX Disruptor的高性能并发框架 项目简介 本项目是一个基于Spring Boot和LMAX Disruptor框架的高性能并发框架,旨在提供高效的事件处理和消息传递机制。项目涵盖了并发编程的核心概念、无锁...

    并发框架Disruptor

    Java并发框架Disruptor,里面采取环形缓存结构,速度更快,适用于生产者消费者模式

    最高效的交易所撮合引擎,采用伦敦外汇交易所LMAX开源的Disruptor框架,用Hazelcast进行分布式内存存.zip

    总的来说,通过结合LMAX的Disruptor并发框架和Hazelcast内存数据网格技术,金融交易所能够构建出高性能的撮合引擎,应对高并发交易需求,提供稳定可靠的服务,同时确保良好的扩展性和容错能力。这不仅是对现有技术的...

    并发编程框架Disruptor.zip

    Disruptor 是一个 Java 的并发编程框架,大大的简化了并发程序开发的难度,在性能上也比 Java 本身提供的一些并发包要好。 标签:Disruptor

    高并发框架Disruptor代码

    《深入解析高并发框架Disruptor》 在现代互联网应用中,处理高并发场景是必不可少的技术挑战。Disruptor,由LMAX公司开发并开源,是一个高性能、低延迟的并发框架,尤其适用于需要大量数据交换的系统。本文将详细...

    Disruptor框架:无锁并发性能的革命

    Disruptor是一个高性能的无锁并发框架,最初由LMAX开发。与传统的基于锁的并发模型相比,Disruptor通过创新的设计大幅提升了数据处理的效率和吞吐量。其核心特性包括环形数组结构、元素位置定位、无锁设计、和多种...

    Disruptor并行框架面试题收录

    Disruptor是一种高效的并发编程框架,主要用于解决高性能系统中的低延迟问题。它通过无锁设计实现了线程间的高效通信与同步,避免了传统锁机制所带来的性能瓶颈。 #### 二、Disruptor核心组件解析 Disruptor框架...

    LMAX.Disruptor,一个无锁高并发框架,中文文档

    LMAX是一种新型零售金融交 易平台,它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上,其核心是一个业务逻辑处理 器,它能够在一个线程里每秒处理6百万订单。...业务逻辑处理器的核心是Disruptor。

    Java工具:高性能并发工具Disruptor简单使用

    Disruptor,由LMAX公司开源的一款并发框架,为处理高并发场景提供了一种新颖且高效的解决方案。它通过消除锁和线程间通信的开销,实现了微秒级的延迟和极高的吞吐量,尤其适用于金融交易、实时分析等对性能有苛刻...

    demo工程列表说明1

    10. **Disruptor并发框架**:disruptor3demo使用了LMAX Disruptor,这是一个高性能并发框架,用于构建低延迟、高吞吐量的应用程序。 11. **Druid数据库连接池**:druiddemo演示了Druid的使用,Druid是阿里巴巴开源...

    Disruptor报错FatalExceptionHandler的解决办法,看网上这种解决办法挺少,整理了一下

    Disruptor是一款高性能的并发框架,它通过使用Ring Buffer和基于事件的处理方式来消除锁竞争,提升系统性能。在使用Disruptor过程中,开发者可能会遇到`FatalExceptionHandler`的错误,这通常是由于处理流程中的异常...

    disruptor-3.2.1源码带jar包20140321

    Disruptor是一款高性能的Java并发框架,由LMAX公司开发并开源。它的设计目标是解决多线程环境下的数据同步问题,特别是在高并发场景下,能够显著提高系统的处理能力。Disruptor的核心设计理念是利用环形缓冲区(Ring...

    Disruptor学习(1)

    《Disruptor学习(1)——深入理解高性能并发框架》 Disruptor是英国LMAX公司开发的一个开源并发框架,其设计目标是为了提供极低延迟的并发处理能力。在金融交易系统等领域,对于高频率、低延迟的需求尤为重要,...

    spring-boot-starter-disruptor.zip

    Disruptor是由LMAX公司开源的一款并发框架,其设计灵感来源于传统的消息队列,但通过独特的环形缓冲区(Ring Buffer)和事件处理机制,显著提升了并发性能,特别适用于高吞吐量、低延迟的场景。Disruptor的核心思想是...

Global site tag (gtag.js) - Google Analytics