序

本文主要研究一下如何使用reactor-rabbitmq

maven

                
     
      io.projectreactor.rabbitmq
                 
     
      reactor-rabbitmq
                 
     
      1.0.0.M2
             
    复制代码

rabbitmq

• 参考
• 当前使用的镜像是bijukunjummen/rabbitmq-server:3.7.0，docker-compose文件配置的账号密码为myuser/mypass
• 访问http://192.168.99.100:15672可以查看界面

实例

@Test    public void testProducer() throws InterruptedException {        int count = 100;        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.useNio();        connectionFactory.setUsername("myuser");        connectionFactory.setPassword("mypass");        SenderOptions senderOptions =  new SenderOptions()                .connectionFactory(connectionFactory)                .connectionSupplier(cf -> cf.newConnection(                        new Address[] {new Address("192.168.99.100",5672), new Address("192.168.99.100",5673), new Address("192.168.99.100",5674)},                        "reactive-sender"))                .resourceCreationScheduler(Schedulers.elastic());        Sender sender = ReactorRabbitMq.createSender(senderOptions);        Flux
    
      confirmations = sender.sendWithPublishConfirms(Flux.range(1, count)                .map(i -> new OutboundMessage("", QUEUE, ("Message_" + i).getBytes())));        CountDownLatch latch = new CountDownLatch(count);        sender.declareQueue(QueueSpecification.queue(QUEUE))                .thenMany(confirmations)                .doOnError(e -> LOGGER.error("Send failed", e))                .subscribe(r -> {                    if (r.isAck()) {                        LOGGER.info("Message {} sent successfully", new String(r.getOutboundMessage().getBody()));                        latch.countDown();                    }                });        latch.await(10, TimeUnit.SECONDS);        sender.close();    }    @Test    public void testConsumer() throws InterruptedException {        int count = 100;        CountDownLatch latch = new CountDownLatch(count);        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.useNio();        connectionFactory.setUsername("myuser");        connectionFactory.setPassword("mypass");        SenderOptions senderOptions =  new SenderOptions()                .connectionFactory(connectionFactory)                .connectionSupplier(cf -> cf.newConnection(                        new Address[] {new Address("192.168.99.100",5672), new Address("192.168.99.100",5673), new Address("192.168.99.100",5674)},                        "reactive-sender"))                .resourceCreationScheduler(Schedulers.elastic());        Sender sender = ReactorRabbitMq.createSender(senderOptions);        Mono
     
       queueDeclaration = sender.declareQueue(QueueSpecification.queue(QUEUE));        ReceiverOptions receiverOptions = new ReceiverOptions()                .connectionFactory(connectionFactory)                .connectionSupplier(cf -> cf.newConnection(                        new Address[] {new Address("192.168.99.100",5672), new Address("192.168.99.100",5673), new Address("192.168.99.100",5674)},                        "reactive-receiver"))                .connectionSubscriptionScheduler(Schedulers.elastic());        Receiver receiver = ReactorRabbitMq.createReceiver(receiverOptions);        Flux
      
        messages = receiver.consumeAutoAck(QUEUE);        Disposable disposable = queueDeclaration.thenMany(messages).subscribe(m -> {            LOGGER.info("Received message {}", new String(m.getBody()));            latch.countDown();        });        latch.await(10, TimeUnit.SECONDS);        disposable.dispose();        sender.close();        receiver.close();    }复制代码
      
     
    

• 由于设置了账号密码，因而需要在ConnectionFactory那里指定账号密码
• 另外由于使用了rabbitmq集群，因而通过connectionSupplier指定要连接的多个rabbitmq地址
• 这里不管是producer还是consumer，都通过queueDeclaration进行操作

小结

reactor-rabbitmq对rabbitmq的api进行封装，改造为reactive streams模式，提供了Non-blocking Back-pressure以及End-to-end Reactive Pipeline特性。

doc