public void publish(String tag, Electron electron) { try { boolean result = super.getEleCircuit().publish(tag, electron); if (!result) { retry(tag, electron); } } catch (Exception e) { if (e instanceof InsufficientCapacityException) { //满了，睡眠300ms重试 try { Thread.sleep(300); } catch (InterruptedException e1) { //ignore } retry(tag, electron); } else { logger.error("Async publish failed electron : { tag: " + tag + "electron: " + electron.toString() + "}", e); } } }
@Override public void addRowChange(RowChange rowChange) { ParamChecker.checkRowChange(tableMeta, rowChange, writerConfig); while (true) { try { long sequence = ringBuffer.tryNext(); RowChangeEvent event = ringBuffer.get(sequence); event.setValue(rowChange); ringBuffer.publish(sequence); return; } catch (InsufficientCapacityException e) { try { Thread.sleep(1); } catch (InterruptedException exp) { } } } }
private void addSignal(ReentrantLock lock, Condition condition) { while (true) { try { long sequence = ringBuffer.tryNext(); RowChangeEvent event = ringBuffer.get(sequence); event.setValue(lock, condition); ringBuffer.publish(sequence); return; } catch (InsufficientCapacityException e) { try { Thread.sleep(1); } catch (InterruptedException exp) { } } } }
public DisruptorQueueImpl(String queueName, ProducerType producerType, int bufferSize, WaitStrategy wait) { this._queueName = PREFIX + queueName; _buffer = RingBuffer.create(producerType, new ObjectEventFactory(), bufferSize, wait); _consumer = new Sequence(); _barrier = _buffer.newBarrier(); _buffer.addGatingSequences(_consumer); if (producerType == ProducerType.SINGLE) { consumerStartedFlag = true; } else { // make sure we flush the pending messages in cache first if (bufferSize < 2) { throw new RuntimeException("QueueSize must >= 2"); } try { publishDirect(FLUSH_CACHE, true); } catch (InsufficientCapacityException e) { throw new RuntimeException("This code should be unreachable!", e); } } }
@Override public void publish(Object obj, boolean block) throws InsufficientCapacityException { boolean publishNow = consumerStartedFlag; if (!publishNow) { readLock.lock(); try { publishNow = consumerStartedFlag; if (!publishNow) { _cache.add(obj); } } finally { readLock.unlock(); } } if (publishNow) { publishDirect(obj, block); } }
/** * @param req * @return */ public boolean processRequest(Runnable req) { long seq; try { seq = m_ringBuffer.tryNext(); } catch (InsufficientCapacityException e1) { return false; } RequestHolder<Runnable> item = (RequestHolder<Runnable>) m_ringBuffer .get(seq); item.put(req); m_ringBuffer.publish(seq); return true; }
/** * @see Sequencer#tryNext(int) */ @Override public long tryNext(int n) throws InsufficientCapacityException { if (n < 1) { throw new IllegalArgumentException("n must be > 0"); } if (!hasAvailableCapacity(n)) { throw InsufficientCapacityException.INSTANCE; } long nextSequence = pad.nextValue += n; return nextSequence; }
/** * @see Sequencer#tryNext(int) */ @Override public long tryNext(int n) throws InsufficientCapacityException { if (n < 1) { throw new IllegalArgumentException("n must be > 0"); } long current; long next; do { current = cursor.get(); next = current + n; if (!hasAvailableCapacity(gatingSequences, n, current)) { throw InsufficientCapacityException.INSTANCE; } } while (!cursor.compareAndSet(current, next)); return next; }
public void publish(Object obj, boolean block) throws InsufficientCapacityException { boolean publishNow = consumerStartedFlag; if (!publishNow) { readLock.lock(); try { publishNow = consumerStartedFlag; if (!publishNow) { _cache.add(obj); } } finally { readLock.unlock(); } } if (publishNow) { publishDirect(obj, block); } }
private ListenableFuture<?> tryPublish(final DOMNotification notification, final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers) { final long seq; try { seq = disruptor.getRingBuffer().tryNext(); } catch (final InsufficientCapacityException e) { return DOMNotificationPublishService.REJECTED; } return publish(seq, notification, subscribers); }
@Override protected Task tryAllocateTask() throws org.camunda.bpm.extension.reactor.projectreactor.processor.InsufficientCapacityException { try { long seqId = ringBuffer.tryNext(); return ringBuffer.get(seqId).setSequenceId(seqId); } catch (InsufficientCapacityException e) { throw org.camunda.bpm.extension.reactor.projectreactor.processor.InsufficientCapacityException.get(); } }
@Override public void publish(Object obj) { try { publish(obj, true); } catch (InsufficientCapacityException ex) { throw new RuntimeException("This code should be unreachable!"); } }
protected void publishDirect(Object obj, boolean block) throws InsufficientCapacityException { final long id; if (block) { id = _buffer.next(); } else { id = _buffer.tryNext(1); } final MutableObject m = _buffer.get(id); m.setObject(obj); _buffer.publish(id); }
public void tryPublish(Object obj) throws InsufficientCapacityException { boolean isSuccess = queue.offer(obj); if (isSuccess == false) { throw InsufficientCapacityException.INSTANCE; } }
@Override public void publish(Object obj, boolean block) throws InsufficientCapacityException { if (block == true) { publish(obj); } else { tryPublish(obj); } }
/** * Attempts to publish an event to the ring buffer. It handles * claiming the next sequence, getting the current (uninitialised) * event from the ring buffer and publishing the claimed sequence * after translation. Will return false if specified capacity * was not available. * * @param translator The user specified translation for the event * @return true if the value was published, false if there was insufficient * capacity. */ public boolean tryPublishEvent(EventTranslator<E> translator) { try { final long sequence = sequencer.tryNext(); translateAndPublish(translator, sequence); return true; } catch (InsufficientCapacityException e) { return false; } }
/** * Allows one user supplied argument. * * @see #tryPublishEvent(EventTranslator) * @param translator The user specified translation for the event * @param arg0 A user supplied argument. * @return true if the value was published, false if there was insufficient * capacity. */ public <A> boolean tryPublishEvent(EventTranslatorOneArg<E, A> translator, A arg0) { try { final long sequence = sequencer.tryNext(); translateAndPublish(translator, sequence, arg0); return true; } catch (InsufficientCapacityException e) { return false; } }
/** * Allows two user supplied arguments. * * @see #tryPublishEvent(EventTranslator) * @param translator The user specified translation for the event * @param arg0 A user supplied argument. * @param arg1 A user supplied argument. * @return true if the value was published, false if there was insufficient * capacity. */ public <A, B> boolean tryPublishEvent(EventTranslatorTwoArg<E, A, B> translator, A arg0, B arg1) { try { final long sequence = sequencer.tryNext(); translateAndPublish(translator, sequence, arg0, arg1); return true; } catch (InsufficientCapacityException e) { return false; } }
/** * Allows three user supplied arguments * * @see #publishEvent(EventTranslator) * @param translator The user specified translation for the event * @param arg0 A user supplied argument. * @param arg1 A user supplied argument. * @param arg2 A user supplied argument. * @return true if the value was published, false if there was insufficient * capacity. */ public <A, B, C> boolean tryPublishEvent(EventTranslatorThreeArg<E, A, B, C> translator, A arg0, B arg1, C arg2) { try { final long sequence = sequencer.tryNext(); translateAndPublish(translator, sequence, arg0, arg1, arg2); return true; } catch (InsufficientCapacityException e) { return false; } }
/** * Allows a variable number of user supplied arguments * * @see #publishEvent(EventTranslator) * @param translator The user specified translation for the event * @param args User supplied arguments. * @return true if the value was published, false if there was insufficient * capacity. */ public boolean tryPublishEvent(EventTranslatorVararg<E> translator, Object...args) { try { final long sequence = sequencer.tryNext(); translateAndPublish(translator, sequence, args); return true; } catch (InsufficientCapacityException e) { return false; } }
/** * Attempts to publish multiple events to the ring buffer. It handles * claiming the next sequence, getting the current (uninitialised) * event from the ring buffer and publishing the claimed sequence * after translation. Will return false if specified capacity * was not available. * * @param translators The user specified translation for the event * @param batchStartsAt The first element of the array which is within the batch. * @param batchSize The actual size of the batch * @return true if all the values were published, false if there was insufficient * capacity. */ public boolean tryPublishEvents(EventTranslator<E>[] translators, int batchStartsAt, int batchSize) { checkBounds(translators, batchStartsAt, batchSize); try { final long finalSequence = sequencer.tryNext(batchSize); translateAndPublishBatch(translators, batchStartsAt, batchSize, finalSequence); return true; } catch (InsufficientCapacityException e) { return false; } }
/** * Allows one user supplied argument. * * @param translator The user specified translation for each event * @param batchStartsAt The first element of the array which is within the batch. * @param batchSize The actual size of the batch * @param arg0 An array of user supplied arguments, one element per event. * @return true if the value was published, false if there was insufficient * capacity. * @see #tryPublishEvents(EventTranslator[]) */ public <A> boolean tryPublishEvents(EventTranslatorOneArg<E, A> translator, int batchStartsAt, int batchSize, A[] arg0) { checkBounds(arg0, batchStartsAt, batchSize); try { final long finalSequence = sequencer.tryNext(batchSize); translateAndPublishBatch(translator, arg0, batchStartsAt, batchSize, finalSequence); return true; } catch (InsufficientCapacityException e) { return false; } }
/** * Allows two user supplied arguments per event. * * @param translator The user specified translation for the event * @param batchStartsAt The first element of the array which is within the batch. * @param batchSize The actual size of the batch. * @param arg0 An array of user supplied arguments, one element per event. * @param arg1 An array of user supplied arguments, one element per event. * @return true if the value was published, false if there was insufficient * capacity. * @see #tryPublishEvents(EventTranslator[]) */ public <A, B> boolean tryPublishEvents(EventTranslatorTwoArg<E, A, B> translator, int batchStartsAt, int batchSize, A[] arg0, B[] arg1) { checkBounds(arg0, arg1, batchStartsAt, batchSize); try { final long finalSequence = sequencer.tryNext(batchSize); translateAndPublishBatch(translator, arg0, arg1, batchStartsAt, batchSize, finalSequence); return true; } catch (InsufficientCapacityException e) { return false; } }
/** * Allows three user supplied arguments per event. * * @param translator The user specified translation for the event * @param batchStartsAt The first element of the array which is within the batch. * @param batchSize The actual size of the batch. * @param arg0 An array of user supplied arguments, one element per event. * @param arg1 An array of user supplied arguments, one element per event. * @param arg2 An array of user supplied arguments, one element per event. * @return true if the value was published, false if there was insufficient * capacity. * @see #publishEvents(EventTranslator[]) */ public <A, B, C> boolean tryPublishEvents(EventTranslatorThreeArg<E, A, B, C> translator, int batchStartsAt, int batchSize, A[] arg0, B[] arg1, C[] arg2) { checkBounds(arg0, arg1, arg2, batchStartsAt, batchSize); try { final long finalSequence = sequencer.tryNext(batchSize); translateAndPublishBatch(translator, arg0, arg1, arg2, batchStartsAt, batchSize, finalSequence); return true; } catch (InsufficientCapacityException e) { return false; } }
/** * Allows a variable number of user supplied arguments per event. * * @param translator The user specified translation for the event * @param batchStartsAt The first element of the array which is within the batch. * @param batchSize The actual size of the batch. * @param args User supplied arguments, one Object[] per event. * @return true if the value was published, false if there was insufficient * capacity. * @see #publishEvents(EventTranslator[]) */ public boolean tryPublishEvents(EventTranslatorVararg<E> translator, int batchStartsAt, int batchSize, Object[]... args) { checkBounds(args, batchStartsAt, batchSize); try { final long finalSequence = sequencer.tryNext(batchSize); translateAndPublishBatch(translator, batchStartsAt, batchSize, finalSequence, args); return true; } catch (InsufficientCapacityException e) { return false; } }
public void push(DisruptorQueue queue, int num) { for (int i = 0; i < num; i++) { String msg = String.valueOf(Thread.currentThread().getId()) + "@" + i; try { queue.publish(msg, false); } catch (InsufficientCapacityException e) { e.printStackTrace(); } produceNum.incrementAndGet(); System.out.println(Thread.currentThread().getId() + " Publish one :" + i); } }
public void publish(Object obj) { try { publish(obj, true); } catch (InsufficientCapacityException ex) { throw new RuntimeException("This code should be unreachable!"); } }
