Storm redis spout 元组丢失无一例外答案

【问题标题】：storm redis spout tuples lost with no exceptionStorm redis spout 元组丢失无一例外
【发布时间】：2018-09-05 13:42:02
【问题描述】：

我已经设置了风暴拓扑（1 个工作人员），其中 spout（在 java 中）从 redis 出列（使用 blpop）事件并转移到螺栓。但是一个观察结果是当队列超过 200 万并且在storm nimbus/supervisor/zookeeper/worker 日志中没有发现警告/异常时，一些事件没有收到到bolt（在clojure，6-spout 线程，50-bolt 线程）。

在本地，这种情况不会使用虚拟数据进行复制。在集群中没有看到网络延迟/数据包丢失。平均处理延迟为 100 毫秒。如何找到在生产中修复它的原因。

(ns event-processor
  (:import [backtype.storm StormSubmitter LocalCluster]
           java.util.UUID
           storm_jedis.RedisQueueSpout
           )
  (:use [backtype.storm clojure config])
  (:require [clojure.tools.logging :as log])
  (:require [clj-redis.client :as redis])
  (:import (redis.clients.jedis Jedis JedisPool JedisPoolConfig))
  (:gen-class))

(defmacro process-event [tuple]
    (log/info "processing event")
    )

(defbolt execute-ls-closure ["word"] {:prepare true}
  [conf context collector]
  (let [counts (atom {})]

    (bolt
     (execute [tuple]
       (let [
        timestart (. System currentTimeMillis)
        tuple-message (.get (get tuple "message") 0)
        string-to-emit (process-event tuple)
        ]
        (emit-bolt! collector [string-to-emit] :anchor tuple)
        (ack! collector tuple)
        )))))

(defn mk-topology []

  (topology
   ;{"1" (spout-spec sentence-spout)
   {"1" (spout-spec redis-spout :p 6)
                     }
   {"3" (bolt-spec {"1" :shuffle }
                   execute-ls-closure
                   :p 50)
                   }))

(defn run-local! []
  (let [cluster (LocalCluster.)]
    (.submitTopology cluster "word-count" {TOPOLOGY-DEBUG true} (mk-topology))
    (Thread/sleep 10000)
    (.shutdown cluster)
    ))

(defn submit-topology! [name]
  (StormSubmitter/submitTopology
   name
   {TOPOLOGY-DEBUG true
    TOPOLOGY-WORKERS 1}
   (mk-topology)))

(defn -main
  ([]
   (run-local!))
  ([name]
   (submit-topology! name)))

【问题讨论】：

我在调试时发现 spout 正确出列但没有将事件传递给 bolt 并且完全没有异常/警告，并且拓扑消息超时为 1 小时，待处理为 5k，批量为 65k。跨度>

标签： java clojure redis apache-storm

【解决方案1】：

如果它不会过多地减慢您的拓扑，您可以使用 Config.setDebug(true) https://github.com/apache/storm/blob/f2ced23fa4e3f699558663baef4ee582ee148fa2/storm-client/src/jvm/org/apache/storm/Config.java#L1763 启用调试日志记录。

否则，我会尝试在螺栓中添加一些调试日志，并为 Redis spout 启用日志记录，以确定元组是否因 Storm 或 Redis 集成而丢失。

另外我注意到您使用的是旧的 Storm 版本。您可以尝试升级。

【讨论】：