Java >> Java チュートリアル >  >> Tag >> Spring

Spring Integration Java DSL を使用した Rabbit MQ との統合

Spring

私は最近、ラスベガスで開催された Spring One カンファレンス 2016 に参加しました。幸運にも、ソフトウェアの世界で長い間尊敬してきた人たちを近くや遠くから見ることができました。数年前に私の Spring Integration 関連のマイナーな貢献のいくつかを実際にマージした 2 人に個人的に会いました。Gary Russel と Artem Bilan で、しばらく使用していなかった Spring Integration をもう一度見るように促されました。

Spring Integration を使用すると、複雑なエンタープライズ統合シナリオがいかに簡単に見えるかをもう一度思い出しました。 Spring Integration Java ベースの DSL が Spring Integration の傘と Spring Cloud Stream のような高レベルの抽象化に完全に統合されていることを嬉しく思います (私の良き友人であり、このプロジェクトの貢献者である Soby Chacko に感謝します
Soby Chacko)。メッセージ駆動型のシナリオのいくつかはさらに簡単です。

この投稿では、RabbitMQ との非常に単純な統合シナリオを再検討しており、後の投稿で Spring Cloud Stream を使用して再実装します。

2 つのサービスが、RabbitMQ ブローカーを介して相互に通信しているシナリオを考えてみましょう。一方は何らかの作業を生成し、もう一方はこの作業を処理します。

プロデューサー

Work unit の生成/ディスパッチ部分は、Spring Integration Java DSL を使用して次のようにコードで表現できます:

@Configuration
public class WorksOutbound {

    @Autowired
    private RabbitConfig rabbitConfig;

    @Bean
    public IntegrationFlow toOutboundQueueFlow() {
        return IntegrationFlows.from("worksChannel")
                .transform(Transformers.toJson())
                .handle(Amqp.outboundAdapter(rabbitConfig.worksRabbitTemplate()))
                .get();
    }
}

これは非常に読みやすいです。フローは、「worksChannel」と呼ばれるチャネルからメッセージを読み取ることから始まり、メッセージを json に変換し、アウトバウンド チャネル アダプターを使用して RabbitMQ 交換にディスパッチします。さて、メッセージは「worksChannel」と呼ばれるチャネルにどのように到達するのでしょうか – Spring Integration ワールドへのエントリ ポイントであるメッセージング ゲートウェイを介して構成しました –

@MessagingGateway
public interface WorkUnitGateway {
 @Gateway(requestChannel = "worksChannel")
 void generate(WorkUnit workUnit);

}

したがって、Java クライアントが「ワーク ユニット」を rabbitmq にディスパッチしたい場合、呼び出しは次のようになります:

WorkUnit sampleWorkUnit = new WorkUnit(UUID.randomUUID().toString(), definition);
workUnitGateway.generate(sampleWorkUnit);

ここでいくつかのことを調べました。具体的には、Rabbit MQ の構成です。ただし、これはごく普通に実行されており、こちらから入手できます。

消費者

プロデューサーの方針に沿って、コンシューマー フローは、RabbitMQ キューからメッセージを受信し、それをドメイン モデルに変換してから、Spring Integration Java DSL を使用して次のように表現されたメッセージを処理することから始まります。

@Configuration
public class WorkInbound {

    @Autowired
    private RabbitConfig rabbitConfig;

    @Autowired
    private ConnectionFactory connectionFactory;

    @Bean
    public IntegrationFlow inboundFlow() {
        return IntegrationFlows.from(
                Amqp.inboundAdapter(connectionFactory, rabbitConfig.worksQueue()).concurrentConsumers(3))
                .transform(Transformers.fromJson(WorkUnit.class))
                .handle("workHandler", "process")
                .get();
    }
}

コードは直観的である必要があります。上記の workHandler は単純な Java pojo であり、次のようになります。ペイロードをログに記録するという非常に重要な仕事をしています:

@Service
public class WorkHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(WorkHandler.class);

    public void process(WorkUnit workUnit) {
        LOGGER.info("Handling work unit - id: {}, definition: {}", workUnit.getId(), workUnit.getDefinition());
    }
}

本質的にはそれだけです。Spring Integration は、ストレート Java と生の RabbitMQ ライブラリを使用して試みた場合、かなり複雑なコードになる素晴らしい外観を提供します。
Spring Cloud Stream は、このセットアップ全体をさらにシンプルにし、今後の投稿のトピック。

このコード全体を github レポジトリに投稿しましたので、試してみたいと思われる方のためにご紹介します。

Java タグ