Disruptor 入门使用案例

深碍√TFBOYSˉ_ 2022-02-13 15:31 281阅读 0赞

## 流程介绍 ##

有一个消费者，有一个生产者，还有一个生产消费队列

*  生产者：生产消息，将消息放入队列
 *  消费者：消费队列中的消息

## 代码： ##

队列中的消息对象：Message

public class Message { 
        String id;
    
        public String getId() { 
            return id;
        }
        public void setId(String id) { 
            this.id = id;
        }
        @Override
        public String toString() { 
            return "Message{" +
                    "id='" + id + '\'' +
                    '}';
        }  
    }

生产者 MessageProducer

public class MessageProducer { 
    	// disrupt 会调用该方法放入队列
        private final static EventTranslatorVararg<Message> translator = (message, seq, objs) -> message.setId((String) objs[0]);
    
        private final RingBuffer<Message> ringBuffer;
    
        public MessageProducer(RingBuffer<Message> ringBuffer) { 
            this.ringBuffer = ringBuffer;
        }
    
    	// 供外部调用
        public void producerNewMessage(String messageId) { 
            this.ringBuffer.publishEvent(translator, messageId);
        }    
    }

消费者

public class MessageDistributionHandler implements EventHandler<Message> { 
        @Override
        public void onEvent(Message message, long sequence, boolean endOfBatch) throws Exception { 
            System.out.println("分发消息 " + message);
        }
    }

Main.java

public class Main { 
        public static void main(String[] args) { 
        	// 创建 disruptor
            Disruptor<Message> disruptor = new Disruptor<>(
                    Message::new,
                    1024,
                    (ThreadFactory) Thread::new);
            // 设置消费者 
            disruptor.handleEventsWith(new MessageDistributionHandler());
            // 启动 disruptor
            disruptor.start();
            // 创建生产者
            MessageProducer producer = new MessageProducer(disruptor.getRingBuffer());
            // 生产消息
            for(int i = 0; i<10; i++){ 
                producer.producerNewMessage(String.valueOf(i));
            }
            // 关闭 disruptor
            disruptor.shutdown();//该方法会阻塞，直至 disruptor 中所有的事件都被处理；
        }
    }