RabbitMQ 创建交换但不创建队列并且没有订阅者接收消息答案

【问题标题】：RabbitMQ creating exchanges but not queues and no subscriber receives messagesRabbitMQ 创建交换但不创建队列并且没有订阅者接收消息
【发布时间】：2021-10-19 14:56:06
【问题描述】：

我正在使用 Micronaut 构建几个微服务。对服务的要求是它们都必须产生事件，但任何服务都可以选择监听其他服务的事件。我想构建它，以便您可以在本地插入 RabbitMQ，但是在任何环境中运行时，您都可以使用 PaaS 代理。

为了实现这一点，我想为每个微服务配置一个交换。例如。产品微服务将向名为product_events 的交换生成事件。该服务也恰好监听来自支付服务的事件：

broker:
  publish:
    exchange: product_events
  subscribe:
    exchanges:
      - payment_events

发布者将消息发布到自己的交易所：

public void sendToTopic(final String routingKey, final String data) throws BrokerPublishException {
    final String rabbitMqFormattedRoutingKey = routingKey.replaceAll("[^A-Za-z0-9 ]", ".");
    final ConnectionFactory factory = getConnectionFactory();
    try (final Connection connection = factory.newConnection(); final Channel channel = connection.createChannel()) {
        LOG.info("Publishing message on exchange '{}' using routing key '{}' with data: {}", exchange, rabbitMqFormattedRoutingKey, data);
        channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true);
        channel.basicPublish(exchange, rabbitMqFormattedRoutingKey, null, data.getBytes(StandardCharsets.UTF_8));
    } catch (final IOException | TimeoutException e) {
        throw new BrokerPublishException("Failed to publish message", e);
    }
}

当我调用服务控制器时，我可以看到消息已生成并添加到交换中：

Publishing message on exchange 'product_events' using routing key 'product.added with data: {"aggregateId":"product"}

然后我希望消费者在应用程序启动时开始：

@Context // Eager start
@ExecuteOn(TaskExecutors.MESSAGE_CONSUMER) // Decouple message consuming to executor
public class SubscriberService {

    private final BrokerSubscriber subscriber;

    public SubscriberService(final BrokerSubscriber subscriber) {
        this.subscriber = subscriber;
        start();
    }

    private void start() {
        subscriber.receive();
    }
}

订阅的 RabbitMQ 实现如下所示：

public void receive() throws BrokerSubscribeException {
    final ConnectionFactory factory = getConnectionFactory();
    try (final Connection connection = factory.newConnection(); final Channel channel = connection.createChannel()){
        for (final String exchange : exchanges) {
            channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true);
            final String queueName = channel.queueDeclare(UUID.randomUUID().toString(), true, true, true, null).getQueue();
            channel.queueBind(queueName, exchange, "#");

            LOG.info("Waiting for messages on exchange '{}', with bound queue '{}'", exchange, queueName);

            final DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                final String message = new String(delivery.getBody(), "UTF-8");
                LOG.info("Received '{}' with message\n:{}", delivery.getEnvelope().getRoutingKey(), message);
            };
            channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
        }
    } catch (final IOException | TimeoutException e) {
        throw new BrokerSubscribeException("Failed when subscribing to messages", e);
    }
}

订阅在启动时生成以下日志消息：

Waiting for messages on exchange 'payment_events', with bound queue 'e60dc206-f757-49c7-8f3f-fee6b2f3d255'

一切正常，我可以看到在管理控制台中创建的交换：

但是控制台中没有队列，我可以重复使用打印到控制台的 guid 并通过管理控制台创建该队列，而不会出现任何冲突或警告。我的服务应该在彼此的交换中侦听 # 根本没有收到任何消息。

我可以通过管理控制台创建一个队列并将其绑定到交换器并获取发布者放置的消息，但我无法让我的服务执行此操作。

简而言之：

应用程序启动，订阅者创建交换并监听#
产品服务获取操作以添加新产品
产品服务将有关添加产品的消息发布到自己的交换中
支付服务监听产品服务交换并记录收到的消息
什么都没有发生

我在同一个 Micronaut 应用程序中创建到同一个代理的两个连接有什么问题吗？

我真的无法理解我在这里做错了什么。

【问题讨论】：

标签： java rabbitmq micronaut

【解决方案1】：

这很尴尬。

订阅者不应使用 try-with-resources，因为连接和通道将在语句结束时立即关闭。简单地改变了：

try (final Connection connection = factory.newConnection(); final Channel channel = connection.createChannel()){
    // Do stuff
}

进入：

try {
    final Connection connection = factory.newConnection();
    final Channel channel = connection.createChannel();
    // Do stuff
}

【讨论】：