客户端使用Java客户端挂起向RabbitMQ发送消息时挂起

问题描述

我已经以Java的反应方式实现RabbitMQ发布者和使用者，但是我的发布功能挂起了频道。队列本身，声明一个队列并消耗，但是工作正常，我已经使用管理员的管理UI对其进行了测试。尝试发送更多消息时，我不再看到“ 队列声明成功”或“ 传递消息进行交换... ”之类的日志。顺便说一句，我知道在liver（）中不需要declareQueue，但是我添加了它来验证在此特定通道中的通信是否有效。

我的代码是：

@Slf4j
@Component
public class RabbitConfigurator {
    private TasksQueueConfig cfg;
    private ReceiverOptions recopts;
    private List<Address> addresses;
    private Utils.ExceptionFunction<ConnectionFactory,Connection> connsupplier;

    public RabbitConfigurator(TasksQueueConfig cfg) {
        this.cfg = cfg;
        addresses = cfg
                .getHosts()
                .stream()
                .map(Address::new)
                .collect(Collectors.toList());
        connsupplier = cf -> {
            LOG.info("initializing new RabbitMQ connection");
            return cf.newConnection(addresses,"dmTasksProc");
        };
    }

    @Bean
    public ConnectionFactory rabbitMQConnectionFactory() {
        ConnectionFactory cf = new ConnectionFactory();
        cf.setHost(cfg.getHosts().get(0));
        cf.setPort(5672);
        cf.setUsername(cfg.getUsername());
        cf.setPassword(cfg.getpassword());
        return cf;
    }

    @Bean
    public Sender sender(ConnectionFactory connFactory) {
        SenderOptions sendOpts = new SenderOptions()
                .connectionClosingTimeout(Duration.parse(cfg.getConnectionTimeout()))
                .connectionFactory(connFactory)
                .connectionsupplier(connsupplier)
                .connectionSubscriptionScheduler(Schedulers.elastic());
        return RabbitFlux.createSender(sendOpts);
    }

    @Bean
    public Receiver receiver(ConnectionFactory connFactory) {
        ReceiverOptions recopts = new ReceiverOptions()
                .connectionClosingTimeout(Duration.parse(cfg.getConnectionTimeout()))
                .connectionFactory(connFactory)
                .connectionsupplier(connsupplier)
                .connectionSubscriptionScheduler(Schedulers.elastic());
        return RabbitFlux.createReceiver(recopts);
    }

    @Bean
    public Flux<Delivery> deliveryFlux(Receiver receiver) {
        return receiver.consumeAutoAck(cfg.getName(),new ConsumeOptions().qos(cfg.getPrefetchCount()));
    }

    @Bean
    public AmqpAdmin rabbitAmqpAdmin(ConnectionFactory connFactory) {
        return new RabbitAdmin(new CachingConnectionFactory(connFactory));
    }
}

和消费者/发布者：

@Slf4j
@Service
public class TasksQueue implements disposableBean {
    private TasksQueueConfig cfg;
    private ObjectMapper mapper;
    private Flux<Delivery> deliveryFlux;
    private Receiver receiver;
    private Sender sender;

    private disposable consumer;

    public TasksQueue(TasksQueueConfig cfg,AmqpAdmin amqpAdmin,ObjectMapper mapper,Flux<Delivery> deliveryFlux,Receiver receiver,Sender sender) {
        this.cfg = cfg;
        this.mapper = mapper;
        this.deliveryFlux = deliveryFlux;
        this.receiver = receiver;
        this.sender = sender;

        amqpAdmin.declareQueue(new Queue(cfg.getName(),false,false));
        consumer = consume();
    }

    public Mono<Void> deliver(Flux<Task> tasks) {
        var pub = sender.sendWithPublishConfirms(
                        tasks.map(task -> {
                            try {
                                String exchange = "";
                                LOG.debug("delivering message to exchange='{}',routingKey='{}'",exchange,cfg.getName());
                                return new OutboundMessage(exchange,cfg.getName(),mapper.writeValueAsBytes(task));
                            } catch(JsonProcessingException ex) {
                                throw Exceptions.propagate(ex);
                            }
                        }));

        return sender.declareQueue(QueueSpecification.queue(cfg.getName()))
                .flatMap(declareOk -> {
                    LOG.info("queue declare success");
                    return Mono.just(declareOk);
                })
                .thenMany(pub)
                .doOnError(JsonProcessingException.class,ex -> LOG.error("Cannot prepare queue message:",ex))
                .doOnError(ex -> LOG.error("Failed to send task to the queue:",ex))
                .map(res -> {
                    if(res.isAck()) {
                        LOG.info("Message {} sent successfully",new String(res.getoutboundMessage().getBody()));
                        return res;
                    } else {
                        LOG.info("todo");
                        return res;
                    }
                })
                .then();
    }

    private disposable consume() {
        return deliveryFlux
                .retrywhen(Retry.fixedDelay(10,Duration.ofSeconds(1)))
                .doOnError(err -> {
                    LOG.error("tasks consumer error",err);
                })
                .subscribe(m -> {
                    LOG.info("Received message {}",new String(m.getBody()));
                });
    }

    @Override
    public void destroy() throws Exception {
        LOG.info("Cleaning up tasks queue resources");
        consumer.dispose();
        receiver.close();
        sender.close();
    }
}

尝试发送消息五分钟后，我得到日志：

r.r.ChannelCloseHandlers$SenderChannelCloseHandler:47: closing channel 1 by signal cancel
r.r.ChannelCloseHandlers$SenderChannelCloseHandler:53: Channel 1 didn't close normally: null

非常感谢您的提前输入！

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）