使用 Reactor 和 RabbitMQ 进行反应式编程

Question

最近我写了一个演示程序来启动反应式编程，结合 Reactor 和 RabbitMQ。这是我的演示代码：

public class FluxWithRabbitMQDemo {

private static final String QUEUE = "demo_thong";

private final reactor.rabbitmq.Sender sender;

private final Receiver receiver;

public FluxWithRabbitMQDemo() {

    this.sender = ReactorRabbitMq.createSender();

    this.receiver = ReactorRabbitMq.createReceiver();

}

public void run(int count) {

    ConnectionFactory connectionFactory = new ConnectionFactory();

    connectionFactory.useNio();

    SenderOptions senderOptions =  new SenderOptions()

            .connectionFactory(connectionFactory)

            .resourceCreationScheduler(Schedulers.elastic());

    reactor.rabbitmq.Sender sender = ReactorRabbitMq.createSender(senderOptions);

    Mono<AMQP.Queue.DeclareOk> queueDeclaration = sender.declareQueue(QueueSpecification.queue(QUEUE));

    Flux<Delivery> messages = receiver.consumeAutoAck(QUEUE);

    queueDeclaration.thenMany(messages).subscribe(m->System.out.println("Get message "+ new String(m.getBody())));

    Flux<OutboundMessageResult> dataStream = sender.sendWithPublishConfirms(Flux.range(1, count)

            .filter(m -> !m.equals(10))

            .parallel()

            .runOn(Schedulers.parallel())

            .doOnNext(i->System.out.println("Message  " + i + " run on thread "+Thread.currentThread().getId()))

            .map(i -> new OutboundMessage("", QUEUE, ("Message " + i).getBytes())));

    sender.declareQueue(QueueSpecification.queue(QUEUE))

            .thenMany(dataStream)

            .doOnError(e -> System.out.println("Send failed"+ e))

            .subscribe(m->{

                if (m!= null){

                    System.out.println("Sent successfully message "+new String(m.getOutboundMessage().getBody()));

                }

            });

    try {

        Thread.sleep(20000);

    } catch (InterruptedException e) {

        e.printStackTrace();

    }

}

我希望在 Flux 发出一个项目后，Sender 必须将它发送到 RabbitMQ，并且在接收到 RabbitMQ 之后，Receiver 必须接收它。但一切都是按顺序发生的，这就是我得到的结果

Answer 1

消息生成太快。要查看交错，请dataStream添加.doOnNext(i->Thread.sleep(10))