public void startProcessing() {
        this.exceptionListener = siddhiAppContext.getRuntimeExceptionListener();
        if (!receivers.isEmpty() && async) {
            for (Constructor constructor : Disruptor.class.getConstructors()) {
                if (constructor.getParameterTypes().length == 5) {      // If new disruptor classes available
                    ProducerType producerType = ProducerType.MULTI;
                    disruptor = new Disruptor<EventExchangeHolder>(
                            new EventExchangeHolderFactory(streamDefinition.getAttributeList().size()),
                            bufferSize, executorService, producerType,
                            new BlockingWaitStrategy());
                    disruptor.handleExceptionsWith(siddhiAppContext.getDisruptorExceptionHandler());
                    break;
                }
            }
            if (disruptor == null) {
                disruptor = new Disruptor<EventExchangeHolder>(
                        new EventExchangeHolderFactory(streamDefinition.getAttributeList().size()),
                        bufferSize, executorService);
                disruptor.handleExceptionsWith(siddhiAppContext.getDisruptorExceptionHandler());
            }
            if (workers > 0) {
                for (int i = 0; i < workers; i++) {
                    disruptor.handleEventsWith(new StreamHandler(receivers, batchSize, streamDefinition.getId(),
                            siddhiAppContext.getName(), faultStreamJunction, onErrorAction, exceptionListener));
                }
            } else {
                disruptor.handleEventsWith(new StreamHandler(receivers, batchSize, streamDefinition.getId(),
                        siddhiAppContext.getName(), faultStreamJunction, onErrorAction, exceptionListener));
            }
            ringBuffer = disruptor.start();
        } else {
            for (Receiver receiver : receivers) {
                if (receiver instanceof StreamCallback) {
                    ((StreamCallback) receiver).startProcessing();
                }
            }
        }
    }
Create and start disruptor based on annotations given in the streamDefinition.
modules/siddhi-core/src/main/java/io/siddhi/core/stream/StreamJunction.java