java - RabbitMQ 消费者未收到消息

标签 java rabbitmq

我正在使用 RabbitMQ 使用 Java 进行工作。
我有两台RabbitMQ服务器,配置相同,一台是开发环境,一台是生产环境。
这是消费者声明:

    /*
     * Connection and channel declaration
     */
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUri(prop.getProperty("ConnectionURI"));
    connection = factory.newConnection();
    channel = connection.createChannel();

    /*
     * Queue declaration and exchange binding
     */
    channel.exchangeDeclare(prop.getProperty("printExchange"), "topic", false, false, false, new HashMap<>());
    queueName = prop.getProperty("printQueue");
    routing_key = "print." + codCliente + "." + idCassa;
    channel.queueDeclare(queueName, false, false, false, null);
    channel.queueBind(queueName, prop.getProperty("printExchange"), routing_key);

这里它开始监听队列:

JAyronPOS.LOGGER.info("Waiting for a message on the queue -> " + queueName + " with routingkey -> " + routing_key);
Consumer consumer = new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        JAyronPOS.LOGGER.info("This is the received message -> " + queueName + ": " + new String(body, "UTF-8"));
        Map<String, Object> headers = properties.getHeaders();
        if (envelope.getRoutingKey().equals(routing_key)) {
            JAyronPOS.LOGGER.info("Message is for me, because it has my routing key");
            channel.basicAck(envelope.getDeliveryTag(), false);
            if (headers != null) {
                if (headers.containsKey("command")) {
                    JAyronPOS.LOGGER.info("It's a command!");
                    JAyronPOS.LOGGER.info(headers.get("command").toString());
                    if ("requestClose".equals(headers.get("command").toString())) {
                        ChiusuraFiscaleConfirm confirm = gson.fromJson(new String(body, "UTF-8"), ChiusuraFiscaleConfirm.class);
                        if (confirm.isCanClose()) {
                            eseguiChiusuraFiscale();
                        } else {
                            JOptionPane.showMessageDialog(null, "Can't close", "Error", JOptionPane.ERROR_MESSAGE);
                        }
                    } else {
                        JAyronPOS.LOGGER.info("Can't handle the message");
                    }
                }
            } else {
                System.out.println("It's a ticket");
                TicketWrapper ticket = gson.fromJson(new String(body, "UTF-8"), TicketWrapper.class);
                printTicket(ticket);
            }
        }else{
            JAyronPOS.LOGGER.info("The message isn't for me, because it has the routingkey: "+envelope.getRoutingKey());
        }
    }
};
channel.basicConsume(queueName, false, consumer);

在开发环境中,我最多有 5 个队列,而在生产环境中,我有 150-200 个队列。
消息由交换器使用个人routing_key发送。发送消息数不高(压力下不超过10条消息/秒)。
当我在开发环境上测试消费者时,一切正常:
- 我发送一个 RPC 调用,服务器处理它并回复。消费者阅读回复并调用正确的方法。大约 1-2 秒内完成。
当我在生产环境中使用该软件时(我仅通过注释/取消注释 config.properties 文件中的一行来更改环境),它不起作用:
- 我发送 RPC 调用,服务器处理它并将回复发送到队列中。消费者永远不会收到消息(但我可以看到 Web 管理面板在队列上开发的消息)。

可能是哪个问题?

编辑:我注意到,如果我发送 RPC 调用,在 RabbitMQ Web 面板中,回复队列上的“Deliver”(浅蓝色)下会有一条消息,而如果我发送 3-4 个 RPC 调用(与前一个相同),在一些调用之后,回复队列上的 Publish(黄色)下会有一条消息,消费者会收到回复。

最佳答案

您尚未提供发布代码或拓扑,因此很难猜测您的拓扑如何工作。

但是在消费者上匹配路由 key 是个坏主意,交易所应该为你做这件事。消费者可以为所需的路由键创建和绑定(bind)队列。 在消费者的代码 else 分支中,不要确认或拒绝消息,这可能会导致消息挂起在 delivered 状态,并且永远不会被其他消费者接收。

Rabbitmq 教程有有用的 RPC 部分 https://www.rabbitmq.com/tutorials/tutorial-six-java.html

关于java - RabbitMQ 消费者未收到消息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/37212713/

相关文章:

java - 将更改推送到 openshift 服务器时构建失败

java - 一种节省空间的数据结构,用于存储和查找大量(均匀分布的)整数

RabbitMQ 从终端声明交换 - 访问被拒绝 :/api/exchanges/

java - 为 rabbitmq 配置 HAProxy

java - 我们需要为 Spring Boot 中的所有内置类创建 bean 吗?

java - 链接 HashMap 按值查找键?

java - 如何修改 JTextPane 中的字母间距?

java - Java 中的重写方法

rabbitmq - Celery 任务计划(Celery、Django 和 RabbitMQ)

java - 超出最大许可数