我构建了一个storm拓扑,它通过一个kafka喷口从ApacheKafka接收元组,将这个数据(使用另一个bolt)作为字符串写入本地系统上的一个.txt文件,然后从我的postbolt发送一个httppost。
两个螺栓都连接到Kafka喷口。
如果我在没有postbolt的情况下测试拓扑,一切正常。但是如果我把螺栓加到拓扑上,整个拓扑就会因为某种原因被阻塞。
有没有人有同样的问题或会给我一个提示,这是什么原因?
我读到closeablehttpclient或closeablehttpresponse阻止线程工作的一些问题。。。在这种情况下,这可能是同一个问题吗?
我的邮政编码:
public class PostBolt extends BaseRichBolt {
private CloseableHttpClient httpclient;
@Override
public final void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
//empty for now
}
@Override
public final void execute(Tuple tuple) {
//create HttpClient:
httpclient = HttpClients.createDefault();
String url = "http://xxx.xxx.xx.xxx:8080/HTTPServlet/httpservlet";
HttpPost post = new HttpPost(url);
post.setHeader("str1", "TEST TEST TEST");
try {
CloseableHttpResponse postResponse;
postResponse = httpclient.execute(post);
System.out.println(postResponse.getStatusLine());
System.out.println("=====sending POST=====");
HttpEntity postEntity = postResponse.getEntity();
//do something useful with the response body
//and ensure that it is fully consumed
EntityUtils.consume(postEntity);
postResponse.close();
}catch (Exception e){
e.printStackTrace();
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("HttpPost"));
}}
我的拓扑代码:
public static void main(String[] args) throws Exception {
/**
* create a config for Kafka-Spout (and Kafka-Bolt)
*/
Config config = new Config();
config.setDebug(true);
config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
//setup zookeeper connection
String zkConnString = "localhost:2181";
//define Kafka topic for the spout
String topic = "mytopic";
//assign the zookeeper connection to brokerhosts
BrokerHosts hosts = new ZkHosts(zkConnString);
//setting up spout properties
SpoutConfig kafkaSpoutConfig = new SpoutConfig(hosts, topic, "/" +topic, UUID.randomUUID().toString());
kafkaSpoutConfig.bufferSizeBytes = 1024 * 1024 * 4;
kafkaSpoutConfig.fetchSizeBytes = 1024 * 1024 * 4;
kafkaSpoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
/**
* Build the Topology by linking the spout and bolts together
*/
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("kafka-spout", new KafkaSpout(kafkaSpoutConfig));
builder.setBolt("printer-bolt", new PrinterBolt()).shuffleGrouping("kafka-spout");
builder.setBolt("post-bolt", new PostBolt()).shuffleGrouping("kafka-spout");
/**
* Check if we're running locally or on a real cluster
*/
if (args != null && args.length >0) {
config.setNumWorkers(6);
config.setNumAckers(6);
config.setMaxSpoutPending(100);
config.setMessageTimeoutSecs(20);
StormSubmitter.submitTopology("StormKafkaTopology", config, builder.createTopology());
} else {
config.setMaxTaskParallelism(3);
config.setNumWorkers(6);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("StormKafkaTopology", config, builder.createTopology());
//Utils.sleep(100000);
//cluster.killTopology("StormKafkaTopology");
//cluster.shutdown();
}
}}
2条答案
按热度按时间92vpleto1#
好吧,我是根据这个评论来确定问题的https://stackoverflow.com/a/32080845/7208987
Kafka喷口将继续重新发送元组,它们没有被发送到的“端点”确认。
所以我只需要确认bolt中传入的元组,拓扑结构的hickup就消失了。
(我发现了这个问题,因为printerbolt确实在继续写,即使kafkaspout没有进一步的输入)。
pw136qt22#
在我看来你已经回答了你的问题但是是的。。。根据这个答案,您应该使用poolighttpclientconnectionmanager,因为您将在多线程环境中运行。
编辑: