Shulou(Shulou.com)05/31 Report--

This article mainly explains "how to use IBasic Bolt to achieve automatic confirmation". The content in the article is simple and clear, and it is easy to learn and understand. Please follow the editor's train of thought to study and learn "how to use IBasic Bolt to achieve automatic confirmation".

Bolt lifecycle

Bolt is a component that takes tuples as input and then produces new tuples as output. When implementing a bolt, you usually need to implement the IRichBolt interface. The Bolts object is created by the client machine, serialized into a topology, and submitted to the hosts in the cluster. The cluster then starts the worker process to deserialize bolt, calls prepare, and finally starts processing tuples.

NOTE: to create a bolt object that initializes member properties with constructor parameters, these property values are serialized when the bolt is submitted to the cluster.

Bolt structure

Bolts has the following methods:

DeclareOutputFields (OutputFieldsDeclarer declarer) {declare output mode for bolt} prepare (java.util.Map stormConf, TopologyContext context, OutputCollector collector) {call} execute (Tuple input) {process input single tuple} cleanup () {when bolt is about to shut down} only before bolt starts processing tuples

Let's look at an example, in which bolt divides a sentence into a list of words:

Class SplitSentence implements IRichBolt {private OutputCollector collector; publlic void prepare (Map conf, TopologyContext context, OutputCollector collector) {this.collector = collector;} public void execute (Tuple tuple) {String sentence = tuple.getString (0); for (String word: sentence.split (")) {collector.emit (new Values (word)) }} public void cleanup () {} public void declareOutputFields (OutputFieldsDeclarer declarer) {declarer.declare (new Fields ("word"));}}

As you can see, this is a very simple bolt. It is worth mentioning that in this case, there is no news guarantee. This means that if bolt discards some message for some reason-- whether because the bolt is dead or because the program deliberately discarded it-- the spout that generated the message will not receive any notification, nor will any other spouts and bolts.

In many cases, however, you want to make sure that the message is processed throughout the topology.

Reliable bolts and unreliable bolts

As mentioned earlier, Storm guarantees that every message sent through spout will be fully processed by all bolt. Based on design considerations, this means that you have to decide for yourself whether your bolts guarantees this.

The topology is a tree structure in which messages (tuples) pass through one or more branches. Each node on the tree calls ack (tuple) or fail (tuple), so Storm knows whether a message has failed and notifies that / the spout (s) that made the messages. Since a Storm topology runs in a highly parallel environment, the best way to track originating spout instances is to include an originating spout reference in the message tuple. This technique is called anchoring (Anchoring). Modify the SplitSentence just mentioned so that it ensures that the messages are processed.

Class SplitSentence implenents IRichBolt {private OutputCollector collector; public void prepare (Map conf, TopologyContext context, OutputCollector collector) {this.collector = collector;} public void execute (Tuple tuple) {String sentence = tuple.getString (0); for (String word: sentence.split (")) {collector.emit (tuple, new Values (word));} collector.ack (tuple) } public void cleanup () {} public void declareOutputFields (OutputFieldsDeclarer declarer) {declar.declare (new Fields ("word"));}}

Anchoring occurs when collector.emit () is called. As mentioned earlier, Storm can trace the originating spout along the tuple. Collector.ack (tuple) and collector.fail (tuple) tell spout what happened to each message. When every message on the tree has been processed, Storm thinks that the tuple from spout is fully processed. If a tuple does not finish processing the message tree within the set timeout period, the tuple processing is considered to have failed. The default timeout is 30 seconds.

NOTE: you can modify the timeout of the topology by modifying the Config.TOPOLOGY_MESSAGE_TIMEOUT.

Of course, spout needs to consider the failure of the message and retry or discard the message accordingly.

NOTE: every message you process is either collector.ack () or collector.fail (). Storm uses memory to track each tuple, so if you don't call these two methods, the task will eventually run out of memory.

Multiple data stream

A bolt can use emit (streamId, tuple) to distribute tuples to multiple streams, where the parameter streamId is a string that identifies the stream. You can then decide which stream to subscribe to it on TopologyBuilder.

Multiple anchoring

In order to connect or aggregate data streams with bolt, you need to use memory buffering tuples. To ensure that the message is completed in this scenario, you have to anchor the flow to multiple tuples. You can do this by passing a list of tuples to the emit method.

List anchors = new ArrayList (); anchors.add (tuple1); anchors.add (tuple2); collector.emit (anchors, values);

In this way, bolt notifies the message tree when it calls the ack or fail method at any time, and because the flow anchors multiple tuples, all relevant spout are notified.

Use IBasicBolt to automatically confirm

As you may have noticed, message confirmation is required in many cases. For simplicity, Storm provides another interface to implement bolt, IBasicBolt. For objects of the implementation class of the interface, the ack method is automatically called after the execute method is executed.

Class SplitSentence extends BaseBasicBolt {public void execute (Tuple tuple, BasicOutputCollector collector) {String sentence = tuple.getString (0); for (String word: sentence.split (")) {collector.emit (new Values (word));}} public void declareOutputFields (OutputFieldsDeclarer declarer) {declarer.declare (new Fields (" word "));}}

NOTE: the BasicOutputCollector that distributes the message is automatically anchored to the tuple passed as a parameter.

Thank you for your reading, the above is "how to use IBasic Bolt to achieve automatic confirmation" content, after the study of this article, I believe you have a deeper understanding of how to use IBasic Bolt to achieve automatic confirmation of this problem, the specific use of the situation also needs to be verified in practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.