Shulou(Shulou.com)06/01 Report--

This article mainly introduces the Disruptor high-performance thread messaging concurrent framework example analysis, has a certain reference value, interested friends can refer to, I hope you can learn a lot after reading this article, the following let the editor with you to understand.

Antecedent words

Disruptor is an open source high-performance concurrency framework for transferring messages between threads of LMAX company in UK. It is very similar to BlockingQueue in jdk, but its performance is incomparable to that of BlockingQueue. The following is an official test report, which can directly see the performance difference between the two:

Core concept?

Such a performance cracked framework must be played with. Before we try it, we first understand the main concepts of disruptor, and then combine the landlord's weblog project (previously used BlockingQueue) to put it into practice.

RingBuffer: a circular buffer, the carrier of message event information. RingBuffer used to be the most important object in Disruptor, but since version 3. 0, its responsibility has been reduced to just storing and updating data (events) exchanged through Disruptor. In some more advanced application scenarios, Ring Buffer can be completely replaced by a user's custom implementation.

Event: defines the types of data that producers and consumers exchange.

EventFactory: a factory class interface for creating events, which is implemented by users and provides specific events

EventHandler: event handling interface, implemented by the user, for handling events.

So far, we can understand the above core content. For more details, you can go to the wiki document: https://github.com/LMAX-Exchange/disruptor

Core architecture diagram:

Practice Disruptor

The transformation of the boot-websocket-log project is an example of a typical producer-consumer model. Then replace BlockingQueue with Disruptor to complete the function, which can be compared if you are interested.

The first step is to define the event type

/ * Created by kl on 2018-8-24 * Content: process log event content carrier * / public class LoggerEvent {private LoggerMessage log; public LoggerMessage getLog () {return log;} public void setLog (LoggerMessage log) {this.log = log;}}

Second, define the event factory

/ * Created by kl on 2018-8-24 * Content: process log event factory class * / public class LoggerEventFactory implements EventFactory {@ Override public LoggerEvent newInstance () {return new LoggerEvent ();}}

Third, define the data processor

/ * Created by kl on 2018-8-24 * Content: process log event handler * / @ Componentpublic class LoggerEventHandler implements EventHandler {@ Autowired private SimpMessagingTemplate messagingTemplate; @ Override public void onEvent (LoggerEvent stringEvent, long l, boolean b) {messagingTemplate.convertAndSend ("/ topic/pullLogger", stringEvent.getLog ());}}

The fourth step is to create the Disruptor action class, define the event publishing method, and publish the event.

/ * Created by kl on 2018-8-24 * Content: Disruptor ring queue * / @ Componentpublic class LoggerDisruptorQueue {private Executor executor = Executors.newCachedThreadPool (); / / The factory for the event private LoggerEventFactory factory = new LoggerEventFactory (); private FileLoggerEventFactory fileLoggerEventFactory = new FileLoggerEventFactory (); / / Specify the size of the ring buffer, must be power of 2.private int bufferSize = 2 * 1024; / / Construct the Disruptor private Disruptordisruptor = new Disruptor (factory, bufferSize, executor); private DisruptorfileLoggerEventDisruptor = new Disruptor (fileLoggerEventFactory, bufferSize, executor); private static RingBufferringBuffer Private static RingBufferfileLoggerEventRingBuffer; @ Autowired LoggerDisruptorQueue (LoggerEventHandler eventHandler,FileLoggerEventHandler fileLoggerEventHandler) {disruptor.handleEventsWith (eventHandler); fileLoggerEventDisruptor.handleEventsWith (fileLoggerEventHandler); this.ringBuffer = disruptor.getRingBuffer (); this.fileLoggerEventRingBuffer = fileLoggerEventDisruptor.getRingBuffer (); disruptor.start (); fileLoggerEventDisruptor.start ();} public static void publishEvent (LoggerMessage log) {long sequence = ringBuffer.next () / / Grab the next sequence try {LoggerEvent event = ringBuffer.get (sequence); / / Get the entry in the Disruptor / / for the sequence event.setLog (log); / / Fill with data} finally {ringBuffer.publish (sequence);}} public static void publishEvent (String log) {if (fileLoggerEventRingBuffer = = null) return; long sequence = fileLoggerEventRingBuffer.next () / / Grab the next sequence try {FileLoggerEvent event = fileLoggerEventRingBuffer.get (sequence); / / Get the entry in the Disruptor / / for the sequence event.setLog (log); / / Fill with data} finally {fileLoggerEventRingBuffer.publish (sequence) Thank you for reading this article carefully. I hope the article "sample Analysis of Disruptor High performance Thread message passing concurrency Framework" shared by the editor will be helpful to you. At the same time, I hope you will support us and pay attention to the industry information channel. More related knowledge is waiting for you to learn!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.