RABBITMQ 向 LOGSTASH 发送消息

Question

你能帮我在 logstash 中输入 RabbitMQ 吗？ 我的应用程序将代码版本发送到 rabbitmq，然后将其存储在弹性堆栈中。 为rabbitmq中的应用程序创建了队列 名称：app_version_queue； 类型：经典； 耐用：真实

然后用该配置配置logstash：

input {
  rabbitmq {
    id => "rabbitmyq_id"
    # connect to rabbit
    host => "localhost"
    port => 5672
    vhost => "/"
    # INPUT - PRODUCERS
    key => "app_version_queue"
    # OUTPUT - CONSUMER
    # queue for logstash
    queue => "logstash"
    auto_delete => false
    # Exchange for logstash
    exchange => logstash
    exchange_type => direct
    durable => "true"
    # No ack will boost your perf
    ack => false
  }
}

output {
  elasticsearch {
    hosts => [ "elasticsearch:9200" ]
    index => "app_version-%{+YYYY.MM.dd}"
  }
}

它有效，但现在，在 RabbitMQ 控制台中，我在表 Queued messages 中看到 准备就绪：914,444 未确认：0 总计：914,444

我在 rabbitmq 集群上的磁盘空间在 3 天内就用完了。 重启rabbitmq服务器后，所有空间都空闲了。

更新： 所有的原因，我为什么这样做，我想从那个链中删除 NIFI app=>rabbit=>nifi=>elastic 我想做：app=>rabbit=>logstash=>elastic

1. 队列：app_version 我的应用程序向 nifi=>ELASTIC 发送消息 Queue1 - app_version_queue
2. 队列：logstash，我用 logstash 创建的 Queue2 - logstash

我尝试停止 NIFI 发送，但消息没有离开。

Answer 1

听起来你已经创建了两次基础架构：

• 在 RabbitMQ 中手动一次
• 在 LogStash 的配置选项中一次

您需要只需三件事：

• 应用程序向其发布消息的交换。
• Lo​​gStash 从中消费消息的队列。
• 该交换与该队列之间的绑定；队列将获得发布到 Exchange 的每条消息的副本，并带有匹配的路由键。

你所拥有的是所有这些：

• 一个名为logs（手动创建）的交换器，您的应用程序将消息发布到该交换器。
• 一个名为 app_version_queue（手动创建）的队列，没有任何消耗。
• 一个绑定（手动创建）将消息副本从logs 传递到app_version_queue，然后永久保存在那里。
• 名为 logstash（由 LogStash 创建）的交换器，没有任何东西向其发布消息。
• 一个名为 logstash（由 LogStash 创建）的队列，LogStash 从中消费消息。
• 从 logstash 交换到 logstash 队列的绑定（由 LogStash 创建）不执行任何操作，因为没有消息发布到该交换。
• 从logs 交换到logstash 队列的绑定（手动创建）实际上是从您的应用程序传递消息。

因此，对于这三件事（交换、队列和绑定）中的每一个，您都需要：

• 决定一个名字
• 决定是创建它还是让 LogStash 创建它
• 将所有内容配置为使用相同的名称

例如，您可以保留名称 logs 和 app_version_queue，并手动创建所有内容。

那么您的 LogStash 应用程序将如下所示：

input {
  rabbitmq {
    id => "rabbitmyq_id"
    # connect to rabbit
    host => "localhost"
    port => 5672
    vhost => "/"
    # Consume from existing queue
    queue => "app_version_queue"
    # No ack will boost your perf
    ack => false
  }
}

另一方面，您可以只创建 logs 交换，让 LogStash 创建队列和绑定，如下所示：

input {
  rabbitmq {
    id => "rabbitmyq_id"
    # connect to rabbit
    host => "localhost"
    port => 5672
    vhost => "/"
    # Create a new queue
    queue => "logstash_processing_queue"
    durable => "true"
    # Take a copy of all messages with the "app_version_queue" routing key from the existing exchange
    exchange => "logs"
    key => "app_version_queue"
    # No ack will boost your perf
    ack => false
  }
}

或者您可以让 LogStash 创建所有它，并确保您的应用程序发布到正确的交换：

input {
  rabbitmq {
    id => "rabbitmyq_id"
    # connect to rabbit
    host => "localhost"
    port => 5672
    vhost => "/"
    # Create a new queue
    queue => "logstash_processing_queue"
    durable => "true"
    # Create a new exchange; point your application to publish here!
    exchange => "log_exchange"
    exchange_type => "direct"
    # Take a copy of all messages with the "app_version_queue" routing key from the new exchange
    key => "app_version_queue"
    # No ack will boost your perf
    ack => false
  }
}

我可能会选择中间选项：交换是应用程序部署要求的一部分（如果它不能在那里发布，它会产生错误），但任何数量的队列可能会出于不同的原因绑定到它（在不需要设置 ElasticSearch 的测试环境中可能根本没有）。