disruptor简单示例

痛定思痛。 2022-08-27 08:53 286阅读 0赞

我们通过一个简单示例来演示disruptor的使用

1.生产者和消费者操作的数据

Person.java

========================

public class Person \{  
    private String name;  
    private int age;  
    private String gender;  
    private String mobile;  
      
    public Person()\{\}  
      
    public Person(String name, int age, String gender, String mobile)\{  
        this.name = name;  
        this.age = age;  
        this.gender = gender;  
        this.mobile = mobile;  
    \}  
      
    public String getName() \{  
        return name;  
    \}  
    public void setName(String name) \{  
        this.name = name;  
    \}  
    public int getAge() \{  
        return age;  
    \}  
    public void setAge(int age) \{  
        this.age = age;  
    \}  
    public String getGender() \{  
        return gender;  
    \}  
    public void setGender(String gender) \{  
        this.gender = gender;  
    \}  
    public String getMobile() \{  
        return mobile;  
    \}  
    public void setMobile(String mobile) \{  
        this.mobile = mobile;  
    \}  
      
      
\}

2.PersonEvent.java

========================

import com.lmax.disruptor.EventFactory;  
/\*\*  
 \* 消费事件  
 \*/  
public class PersonEvent \{  
      
    private Person person;  
  
    public Person getPerson() \{  
        return person;  
    \}  
  
    public void setPerson(Person person) \{  
        this.person = person;  
    \}  
  
    public final static  EventFactory<PersonEvent> EVENT\_FACTORY = new EventFactory<PersonEvent>()\{  
        public PersonEvent newInstance()\{  
            return new PersonEvent();  
        \}  
    \};  
\}

3.消费事件处理处理器

PersonEventHandler.java

========================

import com.lmax.disruptor.EventHandler;  
/\*\*  
 \* 消费事件处理处理器  
 \*/  
public class PersonEventHandler  implements EventHandler<PersonEvent>\{  
      
    public PersonEventHandler()\{  
//        DataSendHelper.start();  
    \}  
    @Override  
    public void onEvent(PersonEvent event, long sequence, boolean endOfBatch)  
            throws Exception \{  
        Person person = event.getPerson();  
        System.out.println("name = "+person.getName()+", age = "+ person.getAge()+", gender = "+ person.getGender() +", mobile = "+person.getMobile());      
    \}  
  
\}

4.ringbuffer的初始化

PersonHelper.java

========================

import com.lmax.disruptor.BatchEventProcessor;  
import com.lmax.disruptor.RingBuffer;  
import com.lmax.disruptor.SequenceBarrier;  
import com.lmax.disruptor.SingleThreadedClaimStrategy;  
import com.lmax.disruptor.YieldingWaitStrategy;  
  
public class PersonHelper \{  
    private  static PersonHelper instance;  
    private static boolean inited = false;  
    /\*\*  
     \* ringBuffer的容量，必须是2的N次方  
     \*/  
    private static final int BUFFER\_SIZE = 256;  
    private RingBuffer<PersonEvent> ringBuffer;  
    private SequenceBarrier sequenceBarrier;  
    private PersonEventHandler handler;  
    private BatchEventProcessor<PersonEvent> batchEventProcessor;  
      
    public PersonHelper()\{  
        //参数1 事件  
        //参数2 单线程使用  
        //参数3 等待策略  
        ringBuffer = new RingBuffer<PersonEvent>(PersonEvent.EVENT\_FACTORY,  
                new SingleThreadedClaimStrategy(BUFFER\_SIZE), new YieldingWaitStrategy());  
        //获取生产者的位置信息  
        sequenceBarrier = ringBuffer.newBarrier();  
        //消费者  
        handler=new PersonEventHandler();  
        //事件处理器，监控指定ringBuffer,有数据时通知指定handler进行处理  
        batchEventProcessor = new BatchEventProcessor<PersonEvent>(ringBuffer, sequenceBarrier, handler);  
        //传入所有消费者线程的序号  
        ringBuffer.setGatingSequences(batchEventProcessor.getSequence());  
          
    \}  
      
    /\*\*  
     \* 启动消费者线程，实际上调用了AudioDataEventHandler中的onEvent方法进行处理  
     \*/  
    public static void start()\{  
        instance = new PersonHelper();  
        Thread thread = new Thread(instance.batchEventProcessor);  
        thread.start();  
        inited = true;  
    \}  
    /\*\*  
     \* 停止  
     \*/  
    public static void shutdown()\{  
         if(!inited)\{  
             throw new RuntimeException("EncodeHelper还没有初始化！");  
         \}else\{  
             instance.doHalt();  
         \}      
    \}  
    private void doHalt() \{  
        batchEventProcessor.halt();  
    \}  
    private void doProduce(Person person)\{  
        //获取下一个序号  
        long sequence = ringBuffer.next();  
        //写入数据  
        ringBuffer.get(sequence).setPerson(person);  
        //通知消费者该资源可以消费了  
        ringBuffer.publish(sequence);  
    \}  
    /\*\*  
     \* 生产者压入生产数据  
     \* @param data  
     \*/  
    public static void produce(Person person)\{  
        if(!inited)\{  
            throw new RuntimeException("EncodeHelper还没有初始化！");  
        \}else\{  
            instance.doProduce(person);  
        \}  
    \}  
      
\}

5. 主函数入口

public class Test \{  
    /\*\*  
     \* @param args  
     \*/  
    public static void main(String\[\] args) \{  
        PersonHelper.start();  
        for(int i=0 ; i<20; i++)\{  
            Person p = new Person("jbgtwang"+i, i , "男", "1234566"+i);

//生产者生产数据  
            PersonHelper.produce(p);  
        \}  
    \}  
  
\}

====================================

**要注意的**：

1.  该示例在电脑上可以正常运行，在android上运行则Unsafe类报错。

网上有人说将disruptor中com.lmax.disruptor.util.Util类中反射调用Unsafe “theUnsafe”的地方改为"THE\_ONE"，重新编译一下就可以用了

2.  ringBuffer的容量，必须是2的N次方