- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
注意:
Bolt1 包含前三个质数(2,3,5)的列表。
Bolt2 包含后三组质数(7,11,13)的列表。
在 Bolt3 中,它只是检查数字是否为素数。
从第一个 bolt 开始,我可以从 spout 类调用 Fail(),但从第二个 bolt 开始,我无法从 spout 类调用 Fail()。
拓扑类:
......
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new SpoutClass(), 1);
builder.setBolt("bolt1", new Bolt1(), 1).shuffleGrouping("spout");
builder.setBolt("bolt2", new Bolt2(), 1).shuffleGrouping("bolt1");
builder.setBolt("bolt3", new Bolt3(), 1).shuffleGrouping("bolt2");
Spout 类:
SpoutClass implements IRichSpout{
private SpoutOutputCollector collector;
private TopologyContext context;
public void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.context = context;
this.collector = collector;
}
public void nextTuple() {
try {
//messageQueue is blocking queue which contains data
String msg = messageQueue.take();
String ackId = msg;
this.collector.emit(new Values(msg), ackId);
}catch (Exception e) {
e.printStackTrace();
}
}
public void ack(Object msgId) {
System.out.println("Acknowledges that this tuple has been processed ........... " + msgId);
}
public void fail(Object msgId) {
System.out.println("FAILED To Process Message :-" + msgId);
}
}
Bolt1 类:
public class Bolt1 extends BaseRichBolt {
private OutputCollector collector;
ArrayList<Integer> firstthreePrime = new ArrayList<Integer>();
firstthreePrime.add(2);
firstthreePrime.add(3);
firstthreePrime.add(5);
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
public void execute(Tuple tuple) {
String message = (String) tuple.getValueByField("msg");
System.out.println("Received " + message + " in Bolt1.");
Integer number = Integer.valueOf(message);
if (check this number contains bolt1 or not) {
//if number is contains
System.out.println(" Number is prime ............." + number + " and Throw from Bolt1");
this.collector.fail(tuple);
} else {
collector.emit(new Values(message));
collector.ack(tuple);
}
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("msg"));
}
}
Bolt2 类:
public class Bolt2 extends BaseRichBolt {
private OutputCollector collector;
ArrayList<Integer> secondthreePrime = new ArrayList<Integer>();
secondthreePrime.add(7);
secondthreePrime.add(11);
secondthreePrime.add(13);
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
public void execute(Tuple tuple) {
String message = (String) tuple.getValueByField("msg");
System.out.println("Received " + message + " in Bolt2.");
Integer number = Integer.valueOf(message);
if (check this number contains bolt2 or not) {
//if number is contains
System.out.println(" Number is prime ............." + number + " and Throw from Bolt2");
this.collector.fail(tuple);
} else {
collector.emit(new Values(message));
collector.ack(tuple);
}
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("msg"));
}
}
Bolt3 类:
public class Bolt3 extends BaseRichBolt {
private OutputCollector collector;
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
public void execute(Tuple tuple) {
String message = (String) tuple.getValueByField("msg");
System.out.println("Received " + message + " in Bolt3.");
Integer number = Integer.valueOf(message);
if (check this number is prime or not) {
//if number is prime
System.out.println(" Number is prime ............." + number + " and Throw from Bolt3");
this.collector.fail(tuple);
} else {
collector.emit(new Values(message));
collector.ack(tuple);
}
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
}
最佳答案
既然您使用的是 BaseRichBolt,您不想锚定传出元组吗?
_collector.emit(tuple, new Values(message));
如果你不锚定它们,那么它们就没有到来自 spout 的元组的链接。查看文档:Guaranteeing Message Processing
关于java - 我在 spout 类中的失败方法仅适用于第一个 bolt ,从第二个 bolt 开始它不起作用。,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45138439/
我在我的 Xcode 项目目录中输入了以下内容: keytool -genkey -v -keystore release.keystore -alias mykey -keyalg RSA \
假设我有一个像这样的 DataFrame(或 Series): Value 0 0.5 1 0.8 2 -0.2 3 None 4 None 5 None
我正在对一个 Pandas 系列进行相对繁重的应用。有什么方法可以返回一些打印反馈,说明每次调用函数时在函数内部进行打印还有多远? 最佳答案 您可以使用跟踪器包装您的函数。以下两个示例,一个基于完成的
我有一个 DataFrame,其中一列包含列表作为单元格内容,如下所示: import pandas as pd df = pd.DataFrame({ 'col_lists': [[1, 2
我想使用 Pandas df.apply 但仅限于某些行 作为一个例子,我想做这样的事情,但我的实际问题有点复杂: import pandas as pd import math z = pd.Dat
我有以下 Pandas 数据框 id dist ds 0 0 0 0 5 1 0 0 7 2 0 0
这发生在我尝试使用 Gradle 构建时。由于字符串是对象,因此似乎没有理由发生此错误: No signature of method: java.util.HashMap.getOrDefault(
您好,有人可以解释为什么在 remaining() 函数中的 Backbone 示例应用程序 ( http://backbonejs.org/examples/todos/index.html ) 中
我有两个域类:用户 class User { String username String password String email Date dateCreated
问题陈述: 一个 pandas dataframe 列系列,same_group 需要根据两个现有列 row 和 col 的值从 bool 值创建。如果两个值在字典 memberships 中具有相似
apporable 报告以下错误: error: unknown type name 'MKMapItem'; did you mean 'MKMapView'? MKMapItem* destina
我有一个带有地址列的大型 DataFrame: data addr 0 0.617964 IN,Krishnagiri,635115 1 0.635428 IN,Chennai
我有一个列表list,里面有这样的项目 ElementA: Number=1, Version=1 ElementB: Number=1, Version=2 ElementC: Number=1,
我正在编译我的源代码,它只是在没有运行应用程序的情况下终止。这是我得到的日志: Build/android-armeabi-debug/com.app4u.portaldorugby/PortalDo
我正在尝试根据另一个单元格的值更改单元格值(颜色“红色”或“绿色”)。我运行以下命令: df.loc[0, 'Colour'] = df.loc[0, 'Count'].apply(lambda x:
我想弄清楚如何使用 StateT结合两个 State基于对我的 Scalaz state monad examples 的评论的状态转换器回答。 看来我已经很接近了,但是在尝试申请 sequence
如果我已经为它绑定(bind)了集合,我该如何添加 RibbonLibrary 默认的快速访问项容器。当我从 UI 添加快速访问工具项时,它会抛出 Operation is not valid whi
在我学习期间Typoclassopedia我遇到了这个证明,但我不确定我的证明是否正确。问题是: One might imagine a variant of the interchange law
我是一名优秀的程序员,十分优秀!