0

堆垛机

我在 RabbitMQ 队列中有很多消息（在我的开发环境中的本地主机上运行）。消息的有效负载是一个 JSON 字符串，我想直接加载到 Elastic Search 中（目前也在 localhost 上运行）。我写了一个快速的 ruby​​ 脚本来从队列中提取消息并将它们加载到 ES 中，如下所示：

#! /usr/bin/ruby
require 'bunny'
require 'json'
require 'elasticsearch'

# Connect to RabbitMQ to collect data
mq_conn = Bunny.new
mq_conn.start
mq_ch = mq_conn.create_channel
mq_q  = mq_ch.queue("test.data")

# Connect to ElasticSearch to post the data
es = Elasticsearch::Client.new log: true

# Main loop - collect the message and stuff it into the db.
mq_q.subscribe do |delivery_info, metadata, payload|
    begin
            es.index index: "indexname",
                     type:  "relationship",
                     body:  payload

    rescue
            puts "Received #{payload} - #{delivery_info} - #{metadata}"
            puts "Exception raised"
            exit
    end
end
mq_conn.close

队列中大约有 4,000,000 条消息。

当我运行脚本时，我看到一堆消息，比如 30 条消息，正在加载到 Elastic Search 中。但是，我看到大约 500 条消息离开队列。 

root@beep:~# rabbitmqctl list_queues
Listing queues ...
test.data    4333080
...done.
root@beep:~# rabbitmqctl list_queues
Listing queues ...
test.data    4332580
...done.

然后脚本静默退出，不告诉我异常。开始/救援块永远不会触发异常，所以我不知道为什么脚本会提前完成或丢失这么多消息。接下来我应该如何调试的任何线索。

一个

4

2 回答 2

1

我在这里添加了一个简单的工作示例：

https://github.com/elasticsearch/elasticsearch-ruby/blob/master/examples/rabbitmq/consumer-publisher.rb

当您不提供测试数据的示例时，很难调试您的示例。

Elasticsearch“河流”功能已被弃用，最终将被删除。如果 RabbitMQ 和 Elasticsearch 是您基础架构的核心部分，您绝对应该花时间编写自己的自定义馈线。

于 2014-03-05T14:31:35.860 回答
0

回答我自己的问题后，我了解到这是将索引指令的消息队列加载到 Elastic 中的一种疯狂而愚蠢的方式。我创建了一条河流，并且可以比使用绳索脚本更快地排出指令。;-)

于 2014-03-05T09:54:54.133 回答