У меня есть носик для выдачи какого-либо запроса, но я получил ошибку, что SpoutOutputCollector не может быть сериализован.Это кажется невозможным.Сообщение об исключении:
java.lang.RuntimeException: java.lang.RuntimeException: java.io.NotSerializableException: org.apache.storm.spout.SpoutOutputCollector
Я пытаюсь войти, прежде чем излучать, и я могу получить журнал.Тем не менее, задача умрет вскоре после сообщения журнала.Вот мой код.
package tiger.server.queryreceiver;
import static tiger.util.Util.QUERY_STREAM;
import com.typesafe.config.Config;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tiger.graph.event.EventTemplate;
import tiger.message.QueryMessage;
import tiger.query.Query;
public abstract class QueryReceiver extends BaseRichSpout {
public static final Logger logger = LoggerFactory.getLogger(QueryReceiver.class);
private static final long serialVersionUID = 5169597660682854052L;
private final EventTemplate template;
private final int queryInterval;
protected SpoutOutputCollector collector;
public QueryReceiver(EventTemplate template, int queryInterval) {
this.template = template;
this.queryInterval = queryInterval;
}
public static QueryReceiver fromConfig(Config config, EventTemplate template) {
//this method will reture the FinalQueryReceiver based on the config
}
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
}
@Override
public void nextTuple() {
List<Query> queries = fetchQueries();
if (queries != null) {
logger.debug("*** emit queries {}", queries);
collector.emit(QUERY_STREAM, new Values(new QueryMessage(queries)));
}
try {
TimeUnit.MILLISECONDS.sleep(queryInterval);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public abstract List<Query> fetchQueries();
}
и класс реализации
package tiger.server.queryreceiver;
import java.util.Arrays;
import java.util.List;
import tiger.graph.event.Event;
import tiger.graph.event.EventTemplate;
import tiger.query.Condition;
import tiger.query.Condition.AlwaysTrueMatcher;
import tiger.query.Condition.SeqMatcher;
import tiger.query.LastEventValueOp;
import tiger.query.NFABuilder;
import tiger.query.Nfa;
import tiger.query.Query;
public class FinalQueryReceiver extends QueryReceiver {
private static final long serialVersionUID = -4372885166586608338L;
private long start;
private final long length;
private final long slide;
private int numSlide;
public FinalQueryReceiver(EventTemplate template, int queryInterval, long start, long length) {
super(template,queryInterval);
this.start = start;
this.length = length;
this.slide = 0;
this.numSlide = 1;
}
public FinalQueryReceiver(EventTemplate template, int queryInterval, long start, long length, long slide, int numSlide) {
super(template, queryInterval);
this.start = start;
this.length = length;
this.slide = slide;
this.numSlide = numSlide;
}
@Override
public List<Query> fetchQueries() {
if(numSlide > 0 || numSlide < 0) {
Nfa nfa = new NFABuilder()
.start(new SeqMatcher() {
@Override
public boolean apply(List<Event> seq, Event event) {
return true;
}
}).op("equal", new LastEventValueOp())
.followedByKleeneEnd(Condition.alwaysTrueMatcher(), Condition.alwaysTrueMatcher())
.op("equal", new LastEventValueOp())
.build();
Query query = new Query(nfa, start, start + length);
start += slide;
if(numSlide > 0) {
numSlide = numSlide - 1;
}
return Arrays.asList(query);
}else {
return null;
}
}
}
Мы можем получить журнал «emit query», однако работник сразу же умрет сразу после журнала
Обновления после некоторых выводов: я запускаю проект в локальном режиме и пытаюсь передать атрибуты queryMessage один за другим, и, наконец, обнаруживаю, что при отправке экземпляра State возникает та же проблема.Код состояния ниже:
public abstract class State implements Serializable {
public Condition.SeqMatcher self;
public Condition.SeqMatcher next;
public String op;
public SeqOperator extractor;
// the rest are all methods and the subclass will only override the methods. the self, next, op, and extractor attributions are all a interface which extend the Serializable.
}
Кроме того, я провел некоторый тест на сериализацию QueryMessage с помощью java ObjectOutputStream, который не показывает исключений.Так что я думаю, что это может быть не проблема сериализации, но в чем действительно проблема?