AbstractStringBuilder.ensureCapacityInternal在风暴之锤得到NullPointerException异常

Question

在线系统，风暴博尔特获得NullPointerException异常，但我认为我行61前检查;偶尔会得到NullPointerException;

import ***.KeyUtils; 
import ***.redis.PipelineHelper; 
import ***.redis.PipelinedCacheClusterClient; 
import **.redis.R2mClusterClient; 
import org.apache.commons.lang3.StringUtils; 
import org.apache.storm.task.OutputCollector; 
import org.apache.storm.task.TopologyContext; 
import org.apache.storm.topology.IRichBolt; 
import org.apache.storm.topology.OutputFieldsDeclarer; 
import org.apache.storm.tuple.Tuple; 
import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
import org.springframework.context.ApplicationContext; 
import org.springframework.context.support.ClassPathXmlApplicationContext; 

import java.util.Map; 

/** 
* RedisBolt batch operate 
*/ 
public class RedisBolt implements IRichBolt { 
    static final long serialVersionUID = 737015318988609460L; 
    private static ApplicationContext applicationContext; 
    private static long logEmitNumber = 0; 
    private static StringBuffer totalCmds = new StringBuffer(); 
    private Logger logger = LoggerFactory.getLogger(getClass()); 
    private OutputCollector _collector; 
    private R2mClusterClient r2mClusterClient; 

    @Override 
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) { 
     _collector = outputCollector; 
     if (applicationContext == null) { 
      applicationContext = new ClassPathXmlApplicationContext("spring/spring-config-redisbolt.xml"); 
     } 
     if (r2mClusterClient == null) { 
      r2mClusterClient = (R2mClusterClient) applicationContext.getBean("r2mClusterClient"); 
     } 


    } 

    @Override 
    public void execute(Tuple tuple) { 
     String log = tuple.getString(0); 
     String lastCommands = tuple.getString(1); 

     try { 
      //log count 
      if (StringUtils.isNotEmpty(log)) { 
       logEmitNumber++; 
      } 

      if (StringUtils.isNotEmpty(lastCommands)) { 
       if(totalCmds==null){ 
        totalCmds = new StringBuffer(); 
       } 
       totalCmds.append(lastCommands);//line 61 
      } 

      //日志数量控制 
      int numberLimit = 1; 
      String flow_log_limit = r2mClusterClient.get(KeyUtils.KEY_PIPELINE_LIMIT); 
      if (StringUtils.isNotEmpty(flow_log_limit)) { 
       try { 
        numberLimit = Integer.parseInt(flow_log_limit); 
       } catch (Exception e) { 
        numberLimit = 1; 
        logger.error("error", e); 
       } 
      } 

      if (logEmitNumber >= numberLimit) { 
       StringBuffer _totalCmds = new StringBuffer(totalCmds); 
       try { 
        //pipeline submit 
        PipelinedCacheClusterClient pip = r2mClusterClient.pipelined(); 
        String[] commandArray = _totalCmds.toString().split(KeyUtils.REDIS_CMD_SPILT); 
        PipelineHelper.cmd(pip, commandArray); 
        pip.sync(); 
        pip.close(); 
        totalCmds = new StringBuffer(); 
       } catch (Exception e) { 
        logger.error("error", e); 
       } 

       logEmitNumber = 0; 
      } 
     } catch (Exception e) { 
      logger.error(new StringBuffer("====RedisBolt error for log=[ ").append(log).append("] \n commands=[").append(lastCommands).append("]").toString(), e); 
      _collector.reportError(e); 
      _collector.fail(tuple); 
     } 

     _collector.ack(tuple); 
    } 

    @Override 
    public void cleanup() { 

    } 

    @Override 
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { 
    } 

    @Override 
    public Map<String, Object> getComponentConfiguration() { 
     return null; 
    } 

} 

异常信息：

显示java.lang.NullPointerException在java.lang.AbstractStringBuilder.ensureCapacityInternal（AbstractStringBuilder.java:113）在java.lang.AbstractStringBuilder.append（AbstractStringBuilder.java:415）在java.lang.StringBuffer.append（StringBuffer.java:237）在com.jd.jr.dataeye.storm.bolt.RedisBolt.execute（RedisBolt.java:61）在org.apache.storm.daemon.executor $ fn__5044 $ tuple_action_fn__5046.invoke（executor.clj：727）在org.apache.storm.daemon.executor $ mk_task_receiver $ fn__4965.invoke（executor.clj：459）在org.apache.storm.disruptor $ clojure_handler $ reify__4480.onEvent（否决。 clj：40）at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor（D isruptorQueue.java:472）在org.apache.storm.utils.DisruptorQueue.consumeBatchWhenAvailable（DisruptorQueue.java:451）在org.apache.storm.disruptor $ consume_batch_when_available.invoke（disruptor.clj：在org.apache.storm 73） .daemon.executor $ fn__5044 $ fn__5057 $ fn__5110.invoke（executor.clj：846）在org.apache.storm.util $ async_loop $ fn__557.invoke（util.clj：484）在clojure.lang.AFn.run（AFN的.java：22）在java.lang.Thread.run（Thread.java:745）

任何人都可以给我一些建议，以找到原因。

Answer 1

这真的很奇怪的事情发生。请阅读两个课程的代码。

https://github.com/openjdk-mirror/jdk7u-jdk/blob/master/src/share/classes/java/lang/AbstractStringBuilder.java

https://github.com/openjdk-mirror/jdk7u-jdk/blob/master/src/share/classes/java/lang/StringBuffer.java

AbstractStringBuilder具有构造与无参数不分配字段“值”，这使得访问“值”字段是NPE。 StringBuffer中的任何构造函数都使用该构造函数。因此，在序列化/反序列化中可能会出现一些奇怪的事情，并且不幸的是AbstractStringBuilder中的'value'字段为空。

也许在prepare（）中初始化totalCmds会更好，还需要考虑螺栓之间的同步（线程安全性）。 prepare（）可以称为每个螺栓实例，因此字段是线程安全的，但类字段不是线程安全的。

Answer 2

我想我可能会发现问题;

关键的一点是

“的StringBuffer _totalCmds =新的StringBuffer（totalCmds）;”和 “totalCmds.append（lastCommands）; //线61”

当新目的，这需要薮步骤：

（1）分配内存，并返回参考

（2）初始化

如果在（1）之后和（2）之前追加，则StringBuffer.java扩展AbstractStringBuilder。java的

/** 
* The value is used for character storage. 
*/ 
char[] value; 

值没有被初始化，那么这将让空：

@Override 
public synchronized void ensureCapacity(int minimumCapacity) { 
    if (minimumCapacity > value.length) { 
     expandCapacity(minimumCapacity); 
    } 
} 

这个印迹有另外一个问题，一些数据的多线程环境