AbstractStringBuilder.ensureCapacityInternal在风暴之锤得到NullPointerException异常
在线系统,风暴博尔特获得NullPointerException异常,但我认为我行61前检查;偶尔会得到NullPointerException;AbstractStringBuilder.ensureCapacityInternal在风暴之锤得到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)
任何人都可以给我一些建议,以找到原因。
这真的很奇怪的事情发生。请阅读两个课程的代码。
AbstractStringBuilder具有构造与无参数不分配字段“值”,这使得访问“值”字段是NPE。 StringBuffer中的任何构造函数都使用该构造函数。因此,在序列化/反序列化中可能会出现一些奇怪的事情,并且不幸的是AbstractStringBuilder中的'value'字段为空。
也许在prepare()中初始化totalCmds会更好,还需要考虑螺栓之间的同步(线程安全性)。 prepare()可以称为每个螺栓实例,因此字段是线程安全的,但类字段不是线程安全的。
我想我可能会发现问题;
关键的一点是
“的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);
}
}
这个印迹有另外一个问题,一些数据的多线程环境
和Java版本下可能失去“1.7 .0_71" – Fanl