Apache Storm java.nio.channels.ClosedChannelException：null

我們正在嘗試使用Apache Storm來處理大量（僞造）消息。留言例如：Apache Storm java.nio.channels.ClosedChannelException：null

"{"clientName":"Sergey Bakulin","sum":12925,"group":"property","suspicious":false,"clientId":2,"dt":1404387303764,"coord":{"lat":55.767842588357645,"lon":37.46920361823332}}".

我們使用Apache卡夫卡作爲我們的風暴簇消息的來源。我們的目標是能夠處理至少50k msg/sec /節點。在當我們使用一個以上的節點，我們一直堅持的錯誤案件（日誌片斷是從工人 - *日誌。）：

2014-07-03 15:14:47 b.s.m.n.Client [INFO] failed to send requests to ip-172-31-23-123.eu-west-1.compute.internal/172.31.23.123:6701: java.nio.channels.ClosedChannelException: null 
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.cleanUpWriteBuffer(AbstractNioWorker.java:381) [netty-3.6.3.Final.jar:na] 
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.close(AbstractNioWorker.java:349) [netty-3.6.3.Final.jar:na] 
at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:93) [netty-3.6.3.Final.jar:na] 
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:107) [netty-3.6.3.Final.jar:na] 
at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312) [netty-3.6.3.Final.jar:na] 
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:88) [netty-3.6.3.Final.jar:na] 
at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) [netty-3.6.3.Final.jar:na] 
at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108) [netty-3.6.3.Final.jar:na] 
at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42) [netty-3.6.3.Final.jar:na] 
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [na:1.7.0_51] 
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [na:1.7.0_51] 
at java.lang.Thread.run(Thread.java:744) [na:1.7.0_51] 
2014-07-03 15:14:47 b.s.m.n.StormClientErrorHandler [INFO] Connection failed Netty-Client-ip-172-31-23-123.eu-west-1.compute.internal/172.31.23.123:6701

我們目前風暴的配置：

########### These MUST be filled in for a storm configuration 
storm.zookeeper.servers: 
    - "172.31.*.*" 

storm.local.dir: "/home/*/storm/data" 
nimbus.host: "127.0.0.1" 
supervisor.slots.ports: 
    - 6701 
    - 6702 

ui.port: 8090 

worker.childopts: "-Xmx6g -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=1%ID% -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun$ 

supervisor.childopts: "-Xmx1024m -Djava.net.preferIPv4Stack=true" 
supervisor.worker.start.timeout.secs: 10 
supervisor.worker.timeout.secs: 10 
supervisor.monitor.frequency.secs: 3 
supervisor.heartbeat.frequency.secs: 5 
supervisor.enable: true 

storm.messaging.netty.server_worker_threads: 2 
storm.messaging.netty.client_worker_threads: 2 
storm.messaging.netty.buffer_size: 5242880 
storm.messaging.netty.max_retries: 25 
storm.messaging.netty.max_wait_ms: 1000

我們的風暴拓撲結構：

Properties conf = Util.readProperties(ClientTopology.class, "storm.properties"); 

prepareRedisDB(conf); 

TopologyBuilder builder = new TopologyBuilder(); 

builder.setSpout("kafka_trans_spout", getKafkaSpout(conf, conf.getProperty("kafka_trans_topic")), 3); 
builder.setSpout("kafka_socevent_spout", getKafkaSpout(conf, conf.getProperty("kafka_socevent_topic")), 3); 

builder.setBolt("json_to_tuple_trans_bolt", new JSONToTupleBolt(Transaction.class), 6) 
     .shuffleGrouping("kafka_trans_spout"); 
builder.setBolt("json_to_tuple_socevent_bolt", new JSONToTupleBolt(SocialEvent.class), 3) 
     .shuffleGrouping("kafka_socevent_spout"); 

builder.setBolt("alert_bolt", new AlertBolt(conf), 3) 
     .fieldsGrouping("json_to_tuple_trans_bolt", new Fields("cl_id")) 
     .fieldsGrouping("json_to_tuple_socevent_bolt", new Fields("cl_id")); 
builder.setBolt("offer_bolt", new NearestOfferBolt(conf), 3) 
     .shuffleGrouping("json_to_tuple_trans_bolt"); 

run(builder, args, 6); 

private static KafkaSpout getKafkaSpout(Properties conf, String topic) { 
    SpoutConfig spoutConfig = new SpoutConfig(
      new ZkHosts(conf.getProperty("zk_host"), "/brokers"), 
      topic, 
      "/brokers", 
      conf.getProperty("kafka_consumer_group_id")); 
    List<String> zkServers = new ArrayList<String>(); 
    zkServers.add(conf.getProperty("zk_host")); 
    spoutConfig.zkServers = zkServers; 
    spoutConfig.zkPort = Integer.valueOf(conf.getProperty("zk_port")); 
    spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme()); 
    spoutConfig.forceFromStart = true; 
    spoutConfig.fetchSizeBytes = 5*1024*1024; 
    spoutConfig.bufferSizeBytes = 5*1024*1024; 
    storm.kafka.KafkaSpout kafkaSpout = new storm.kafka.KafkaSpout(spoutConfig); 
    return kafkaSpout; 
}

我們使用AWS c3.2xlarge機，阿帕奇風暴0.9.2-孵化，阿帕奇卡夫卡2.9.2-0.8.1.1。

來源

2014-07-03 intsco

您可以檢查您確實在172.31.23.123:6701 上有偵聽服務嘗試netstat -antp | grep 6701在這臺機器上 – Pixou

你有解決這個問題嗎？我現在得到同樣的錯誤。 – gjain

異常看起來像主管的端口不能從外部世界訪問。看看這個鏈接：https://gist.github.com/amontalenti/8ff0c31a7b95a6dea3d2 你有沒有嘗試telnet到該主機端口？ – Shams

測試ping和telnet： 確保運行風暴每臺機器都有連接用ping所有其他計算機（所有工人，靈氣和動物園管理員）。嘗試通過IP，主機名和FQDN ping，如果不起作用，請編輯主機（/ etc/hosts）文件。

另外，telnet機器以檢查storm.yaml（6701,6702）中的開放端口。動物園管理員（2181）。

在我的測試環境下，storm.yaml設置適用於以下網狀設置：

storm.messaging.netty.buffer_size: 5242880 
storm.messaging.netty.client_worker_threads: 1 
storm.messaging.netty.max_retries: 100 
storm.messaging.netty.max_wait_ms: 1000 
storm.messaging.netty.min_wait_ms: 100 
storm.messaging.netty.server_worker_threads: 1 
storm.messaging.transport: backtype.storm.messaging.netty.Context

來源

2016-03-21 14:09:28 LiozM

首先嚐試增加負載，然後運行的拓撲結構，它發生幾次與我的主題是新的和負載不存在。

來源

2016-07-11 07:40:02

Apache Storm java.nio.channels.ClosedChannelException：null

回答

相關問題