2

私は Rebus を使用しており、CQRS Journeyの段落「イベントを複数回処理するのを避ける」のようなものを導入したいのですが、わかりません。

トランスポートには SQL Server を使用し、サブスクリプションとサガには MongoDBを使用するように Rebus を構成しました。RoutingはTypeBasedで構成され、すべてのコマンド ハンドラーの型をTransportで構成されたキューにマップします。

 var bus = Configure.With(new SimpleInjectorContainerAdapter(container))
            .Logging(l => l.Trace())
            .Transport(t =>
            {
                t.UseSqlServer(connectionstring, "TestMessages", "messageQueueName");
            })
            .Routing(r => r.TypeBased()
                            .MapAssemblyOf<Assembly1.Commands.DoSomething>("messageQueueName")
                            .MapAssemblyOf<Assembly2.Commands.DoSomethingElse>("messageQueueName")
                             )
            .Sagas(s => s.StoreInMongoDb(db, (sagaType) =>
            {
                return sagaType.Name;
            }))
            .Subscriptions(s => s.StoreInMongoDb(db, "Subscriptions"))
            .Options(o =>
            {
                o.SetNumberOfWorkers(1);
                o.SetMaxParallelism(1);
                o.EnableSagaAuditing().StoreInMongoDb(db, "Snapshots");
            })
            .Start();

ここで、コマンドがイベントを発行するときに、これが既存のサブスクライバーのタイプと同じ数の個別のトピック (仮想または物理キュー) にレプリケートされるように Rebus を構成する必要があります。

何かのようなもの：

bus.Subscribe<Assembly1.EventHandler1>("Assembly1.EventHandler1Queue").Wait();
bus.Subscribe<Assembly1.EventHandler2>("Assembly1.EventHandler2Queue").Wait();
bus.Subscribe<Assembly2.EventHandler1>("Assembly2.EventHandler1Queue").Wait();

手伝ってくれてありがとう。

4

1 に答える 1

2

あなたの質問と紛らわしいと思われることがいくつかあります。

しかし、あなたの基本的な質問は、各メッセージが各サブスクライバーによって 1 回だけ処理されるようにする方法だと思います。

答えは簡単です。各サブスクライバーに個別のエンドポイントを用意します。これは、各サブスクライバーが、メッセージが処理され、失敗したメッセージが返される独自の入力キューを持つことを意味します。

その後、各サブスクライバーに必要な数のハンドラーを含めることができます。受信メッセージごとに、互換性のあるすべてのハンドラーが実行されます。

Rebus を使用すると、呼び出しごとConfigure.With(...).(...).Start()に個別のエンドポイントが提供されます。そのため、この場合、サブスクライバー エンドポイントの作成をメソッドでラップすることをお勧めします。メソッドは次のように呼び出すことができます。

var event1Subscriber = CreateSubscriber("subscriber_event1");
event1Subscriber.Subscribe<Event1>().Wait();

var event2Subscriber = CreateSubscriber("subscriber_event2");
event2Subscriber.Subscribe<Event2>().Wait();

var event3Subscriber = CreateSubscriber("subscriber_event3");    
event3Subscriber.Subscribe<Event3>().Wait();

// ...

CreateSubscriber次に、次のような場所になります。

public IBus CreateSubscriber(string queueName)
{
    return Configure.With(GetContainerAdapter())
        .Transport(t => t.UseMsmq(queueName))
        .Start();        
}
于 2016-03-16T06:45:13.860 に答える