RabbitMQ — Mesaj Hangi Kuyruğa Gider?

1
@Configuration
2
class EventTopology {
3

4
    @Bean
5
    TopicExchange bankEvents() {
6
        return ExchangeBuilder.topicExchange("bank.events")
7
                .durable(true)
8
                .alternate("bank.unroutable")      // nothing matched? keep it, don't drop it
9
                .build();
10
    }
11

12
    @Bean
13
    FanoutExchange unroutableExchange() {
14
        return new FanoutExchange("bank.unroutable");
15
    }
16

17
    @Bean Queue notifyQueue()     { return QueueBuilder.durable("notify").build(); }
18
    @Bean Queue fraudQueue()      { return QueueBuilder.durable("fraud").build(); }
19
    @Bean Queue reportQueue()     { return QueueBuilder.durable("report").build(); }
20
    @Bean Queue unroutableQueue() { return QueueBuilder.durable("unroutable").build(); }
21

22
    @Bean
23
    Declarables bindings(TopicExchange bankEvents, FanoutExchange unroutableExchange,
24
                         Queue notifyQueue, Queue fraudQueue, Queue reportQueue, Queue unroutableQueue) {
25
        return new Declarables(
26
                BindingBuilder.bind(notifyQueue).to(bankEvents).with("odeme.tamamlandi.*"),
27
                BindingBuilder.bind(fraudQueue).to(bankEvents).with("*.basarisiz.*"),
28
                BindingBuilder.bind(fraudQueue).to(bankEvents).with("kart.#"),
29
                BindingBuilder.bind(reportQueue).to(bankEvents).with("#"),
30
                BindingBuilder.bind(unroutableQueue).to(unroutableExchange));
31
    }
32
}

1
@Component
2
class BankEventPublisher {
3

4
    private final RabbitTemplate rabbit;
5

6
    BankEventPublisher(RabbitTemplate rabbit) {
7
        this.rabbit = rabbit;
8
    }
9

10
    /** The publisher knows the exchange and the key, never the queues. */
11
    void cardBlocked(CardBlocked event) {
12
        rabbit.convertAndSend("bank.events", "kart.bloke." + event.region(), event);
13
    }
14

15
    void paymentCompleted(PaymentCompleted event) {
16
        rabbit.convertAndSend("bank.events", "odeme.tamamlandi." + event.region(), event);
17
    }
18
}

1
@Component
2
class FraudListener {
3

4
    private final FraudChecks checks;
5

6
    FraudListener(FraudChecks checks) {
7
        this.checks = checks;
8
    }
9

10
    // With the default AUTO acknowledge mode, Spring acks only after this method returns;
11
    // an exception leaves the message to be redelivered instead of losing it.
12
    @RabbitListener(queues = "fraud")
13
    void on(BankEvent event) {
14
        checks.review(event);
15
    }
16
}

1
@Component
2
class UnroutableWatcher {
3

4
    private final Counter unroutable;
5

6
    UnroutableWatcher(MeterRegistry registry) {
7
        this.unroutable = registry.counter("events.unroutable");
8
    }
9

10
    /** A message here means a routing key nobody is bound to: an alert, not noise. */
11
    @RabbitListener(queues = "unroutable")
12
    void on(Message message) {
13
        unroutable.increment();
14
    }
15
}