0

メッセージを処理するためにディスラプターを使用しようとしています。2 段階の処理が必要です。つまり、次のようなワーカー プールで動作する 2 つのグループのハンドラー (推測):

disruptor.
handleEventsWithWorkerPool(
    firstPhaseHandlers)
.thenHandleEventsWithWorkerPool(
    secondPhaseHandlers);

上記のコードを使用する場合、各グループに複数のワーカーを配置すると、パフォーマンスが低下します。つまり、まったく同じ量の作業に対して大量の CPU が浪費されています。

リング バッファ サイズを微調整しようとしましたが (パフォーマンスに影響があることは既に確認済みです)、この場合は役に立ちませんでした。だから私は何か間違ったことをしていますか、それともこれは本当の問題ですか?

問題の完全なデモを添付しています。

import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;

final class ValueEvent {
private long value;

public long getValue() {
    return value;
}

public void setValue(long value) {
    this.value = value;
}

public final static EventFactory<ValueEvent> EVENT_FACTORY = new     EventFactory<ValueEvent>() {
    public ValueEvent newInstance() {
        return new ValueEvent();
    }
};
}

class MyWorkHandler implements WorkHandler<ValueEvent> {

AtomicLong workDone;
public MyWorkHandler (AtomicLong wd)
{
    this.workDone=wd;
}
public void onEvent(final ValueEvent event) throws Exception {

    workDone.incrementAndGet();
}

}

class My2ndPahseWorkHandler implements WorkHandler<ValueEvent> {


AtomicLong workDone;
public My2ndPahseWorkHandler (AtomicLong wd)
{
    this.workDone=wd;
}

public void onEvent(final ValueEvent event) throws Exception {

    workDone.incrementAndGet();
}

}

class MyEventTranslator implements EventTranslatorOneArg<ValueEvent, Long> {

@Override
public void translateTo(ValueEvent event, long sequence, Long value) {

    event.setValue(value);

}

}

public class TwoPhaseDisruptor {

static AtomicLong workDone=new AtomicLong(0);

@SuppressWarnings("unchecked")
public static void main(String[] args) {

    ExecutorService exec = Executors.newCachedThreadPool();

    int numOfHandlersInEachGroup=Integer.parseInt(args[0]);
    long eventCount=Long.parseLong(args[1]);
    int ringBufferSize=2 << (Integer.parseInt(args[2]));


    Disruptor<ValueEvent> disruptor = new Disruptor<ValueEvent>(
            ValueEvent.EVENT_FACTORY, ringBufferSize,
            exec);

    ArrayList<MyWorkHandler> handlers = new ArrayList<MyWorkHandler>();
    for (int i = 0; i < numOfHandlersInEachGroup ; i++) {

        handlers.add(new MyWorkHandler(workDone));
    }

    ArrayList<My2ndPahseWorkHandler > phase2_handlers = new ArrayList<My2ndPahseWorkHandler >();
    for (int i = 0; i < numOfHandlersInEachGroup; i++) {
        phase2_handlers.add(new My2ndPahseWorkHandler(workDone));
    }

    disruptor
            .handleEventsWithWorkerPool(
                    handlers.toArray(new WorkHandler[handlers.size()]))
            .thenHandleEventsWithWorkerPool(
                    phase2_handlers.toArray(new WorkHandler[phase2_handlers.size()]));

    long s = (System.currentTimeMillis());
    disruptor.start();

    MyEventTranslator myEventTranslator = new MyEventTranslator();
    for (long i = 0; i < eventCount; i++) {
        disruptor.publishEvent(myEventTranslator, i);
    }

    disruptor.shutdown();
    exec.shutdown();
    System.out.println("time spent "+ (System.currentTimeMillis() - s) + "     ms");
    System.out.println("amount of work done "+ workDone.get());
}
}

各グループに 1 つのスレッドで上記の例を実行してみてください

1 100000 7

私のコンピューターでそれが与えた

time spent 371 ms
amount of work done 200000

次に、各グループで 4 つのスレッドで試します

4 100000 7

私のコンピューターで与えたもの

time spent 9853 ms
amount of work done 200000

実行中、CPU の使用率は 100% です

4

1 に答える 1