Google Git
Sign in
apache / storm-site / b711e70434639e86ade5bfc8a72a132abcb3079a / . / content / releases / 2.1.0 / storm-rocketmq.html
blob: 9f2f08d0fe4b43d09d57709ccdd816693416f52d [file] [log] [blame]
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1">
<link rel="shortcut icon" href="/favicon.ico" type="image/x-icon">
<link rel="icon" href="/favicon.ico" type="image/x-icon">
<title>Storm RocketMQ</title>
<!-- Bootstrap core CSS -->
<link href="/assets/css/bootstrap.min.css" rel="stylesheet">
<!-- Bootstrap theme -->
<link href="/assets/css/bootstrap-theme.min.css" rel="stylesheet">
<!-- Custom styles for this template -->
<link rel="stylesheet" href="http://fortawesome.github.io/Font-Awesome/assets/font-awesome/css/font-awesome.css">
<link href="/css/style.css" rel="stylesheet">
<link href="/assets/css/owl.theme.css" rel="stylesheet">
<link href="/assets/css/owl.carousel.css" rel="stylesheet">
<script type="text/javascript" src="/assets/js/jquery.min.js"></script>
<script type="text/javascript" src="/assets/js/bootstrap.min.js"></script>
<script type="text/javascript" src="/assets/js/owl.carousel.min.js"></script>
<script type="text/javascript" src="/assets/js/storm.js"></script>
<!-- Just for debugging purposes. Don't actually copy these 2 lines! -->
<!--[if lt IE 9]><script src="../../assets/js/ie8-responsive-file-warning.js"></script><![endif]-->
<!-- HTML5 shim and Respond.js for IE8 support of HTML5 elements and media queries -->
<!--[if lt IE 9]>
<script src="https://oss.maxcdn.com/html5shiv/3.7.2/html5shiv.min.js"></script>
<script src="https://oss.maxcdn.com/respond/1.4.2/respond.min.js"></script>
<![endif]-->
</head>
<body>
<header>
<div class="container-fluid">
<div class="row">
<div class="col-md-5">
<a href="/index.html"><img src="/images/logo.png" class="logo" /></a>
</div>
<div class="col-md-5">
<h1>Version: 2.1.0</h1>
</div>
<div class="col-md-2">
<a href="/downloads.html" class="btn-std btn-block btn-download">Download</a>
</div>
</div>
</div>
</header>
<!--Header End-->
<!--Navigation Begin-->
<div class="navbar" role="banner">
<div class="container-fluid">
<div class="navbar-header">
<button class="navbar-toggle" type="button" data-toggle="collapse" data-target=".bs-navbar-collapse">
<span class="icon-bar"></span>
<span class="icon-bar"></span>
<span class="icon-bar"></span>
</button>
</div>
<nav class="collapse navbar-collapse bs-navbar-collapse" role="navigation">
<ul class="nav navbar-nav">
<li><a href="/index.html" id="home">Home</a></li>
<li><a href="/getting-help.html" id="getting-help">Getting Help</a></li>
<li><a href="/about/integrates.html" id="project-info">Project Information</a></li>
<li class="dropdown">
<a href="#" class="dropdown-toggle" data-toggle="dropdown" id="documentation">Documentation <b class="caret"></b></a>
<ul class="dropdown-menu">
<li><a href="/releases/2.3.0/index.html">2.3.0</a></li>
<li><a href="/releases/2.2.0/index.html">2.2.0</a></li>
<li><a href="/releases/2.1.0/index.html">2.1.0</a></li>
<li><a href="/releases/2.0.0/index.html">2.0.0</a></li>
<li><a href="/releases/1.2.3/index.html">1.2.3</a></li>
</ul>
</li>
<li><a href="/talksAndVideos.html">Talks and Slideshows</a></li>
<li class="dropdown">
<a href="#" class="dropdown-toggle" data-toggle="dropdown" id="contribute">Community <b class="caret"></b></a>
<ul class="dropdown-menu">
<li><a href="/contribute/Contributing-to-Storm.html">Contributing</a></li>
<li><a href="/contribute/People.html">People</a></li>
<li><a href="/contribute/BYLAWS.html">ByLaws</a></li>
</ul>
</li>
<li><a href="/2021/09/27/storm230-released.html" id="news">News</a></li>
</ul>
</nav>
</div>
</div>
<div class="container-fluid">
<h1 class="page-title">Storm RocketMQ</h1>
<div class="row">
<div class="col-md-12">
<!-- Documentation -->
<p class="post-meta"></p>
<div class="documentation-content"><p>Storm/Trident integration for <a href="https://rocketmq.incubator.apache.org/">RocketMQ</a>. This package includes the core spout, bolt and trident states that allows a storm topology to either write storm tuples into a topic or read from topics in a storm topology.</p>
<h2 id="read-from-topic">Read from Topic</h2>
<p>The spout included in this package for reading data from a topic.</p>
<h3 id="rocketmqspout">RocketMqSpout</h3>
<p>To use the <code>RocketMqSpout</code>, you construct an instance of it by specifying a Properties instance which including rocketmq configs.
RocketMqSpout uses RocketMQ MQPushConsumer as the default implementation. PushConsumer is a high level consumer API, wrapping the pulling details. Looks like broker push messages to consumer.
RocketMqSpout will retry 3(use <code>SpoutConfig.DEFAULT_MESSAGES_MAX_RETRY</code> to change the value) times when messages are failed.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"> <span class="n">Properties</span> <span class="n">properties</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
<span class="n">properties</span><span class="o">.</span><span class="na">setProperty</span><span class="o">(</span><span class="n">SpoutConfig</span><span class="o">.</span><span class="na">NAME_SERVER_ADDR</span><span class="o">,</span> <span class="n">nameserverAddr</span><span class="o">);</span>
<span class="n">properties</span><span class="o">.</span><span class="na">setProperty</span><span class="o">(</span><span class="n">SpoutConfig</span><span class="o">.</span><span class="na">CONSUMER_GROUP</span><span class="o">,</span> <span class="n">group</span><span class="o">);</span>
<span class="n">properties</span><span class="o">.</span><span class="na">setProperty</span><span class="o">(</span><span class="n">SpoutConfig</span><span class="o">.</span><span class="na">CONSUMER_TOPIC</span><span class="o">,</span> <span class="n">topic</span><span class="o">);</span>
<span class="n">RocketMqSpout</span> <span class="n">spout</span> <span class="o">=</span> <span class="k">new</span> <span class="n">RocketMqSpout</span><span class="o">(</span><span class="n">properties</span><span class="o">);</span>
</code></pre></div>
<h2 id="write-into-topic">Write into Topic</h2>
<p>The bolt and trident state included in this package for write data into a topic.</p>
<h3 id="tupletomessagemapper">TupleToMessageMapper</h3>
<p>The main API for mapping Storm tuple to a RocketMQ Message is the <code>org.apache.storm.rocketmq.common.mapper.TupleToMessageMapper</code> interface:</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">interface</span> <span class="nc">TupleToMessageMapper</span> <span class="kd">extends</span> <span class="n">Serializable</span> <span class="o">{</span>
<span class="n">String</span> <span class="nf">getKeyFromTuple</span><span class="o">(</span><span class="n">ITuple</span> <span class="n">tuple</span><span class="o">);</span>
<span class="kt">byte</span><span class="o">[]</span> <span class="nf">getValueFromTuple</span><span class="o">(</span><span class="n">ITuple</span> <span class="n">tuple</span><span class="o">);</span>
<span class="o">}</span>
</code></pre></div>
<h3 id="fieldnamebasedtupletomessagemapper">FieldNameBasedTupleToMessageMapper</h3>
<p><code>storm-rocketmq</code> includes a general purpose <code>TupleToMessageMapper</code> implementation called <code>FieldNameBasedTupleToMessageMapper</code>.</p>
<h3 id="topicselector">TopicSelector</h3>
<p>The main API for selecting topic and tags is the <code>org.apache.storm.rocketmq.common.selector.TopicSelector</code> interface:</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">interface</span> <span class="nc">TopicSelector</span> <span class="kd">extends</span> <span class="n">Serializable</span> <span class="o">{</span>
<span class="n">String</span> <span class="nf">getTopic</span><span class="o">(</span><span class="n">ITuple</span> <span class="n">tuple</span><span class="o">);</span>
<span class="n">String</span> <span class="nf">getTag</span><span class="o">(</span><span class="n">ITuple</span> <span class="n">tuple</span><span class="o">);</span>
<span class="o">}</span>
</code></pre></div>
<h3 id="defaulttopicselector-fieldnamebasedtopicselector">DefaultTopicSelector/FieldNameBasedTopicSelector</h3>
<p><code>storm-rocketmq</code> includes general purpose <code>TopicSelector</code> implementations called <code>DefaultTopicSelector</code> and <code>FieldNameBasedTopicSelector</code>.</p>
<h3 id="rocketmqbolt">RocketMqBolt</h3>
<p>To use the <code>RocketMqBolt</code>, you construct an instance of it by specifying TupleToMessageMapper, TopicSelector and Properties instances.
RocketMqBolt send messages async by default. You can change this by invoking <code>withAsync(false)</code>.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"> <span class="n">TupleToMessageMapper</span> <span class="n">mapper</span> <span class="o">=</span> <span class="k">new</span> <span class="n">FieldNameBasedTupleToMessageMapper</span><span class="o">(</span><span class="s">"word"</span><span class="o">,</span> <span class="s">"count"</span><span class="o">);</span>
<span class="n">TopicSelector</span> <span class="n">selector</span> <span class="o">=</span> <span class="k">new</span> <span class="n">DefaultTopicSelector</span><span class="o">(</span><span class="n">topic</span><span class="o">);</span>
<span class="n">properties</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
<span class="n">properties</span><span class="o">.</span><span class="na">setProperty</span><span class="o">(</span><span class="n">RocketMqConfig</span><span class="o">.</span><span class="na">NAME_SERVER_ADDR</span><span class="o">,</span> <span class="n">nameserverAddr</span><span class="o">);</span>
<span class="n">RocketMqBolt</span> <span class="n">insertBolt</span> <span class="o">=</span> <span class="k">new</span> <span class="n">RocketMqBolt</span><span class="o">()</span>
<span class="o">.</span><span class="na">withMapper</span><span class="o">(</span><span class="n">mapper</span><span class="o">)</span>
<span class="o">.</span><span class="na">withSelector</span><span class="o">(</span><span class="n">selector</span><span class="o">)</span>
<span class="o">.</span><span class="na">withProperties</span><span class="o">(</span><span class="n">properties</span><span class="o">);</span>
</code></pre></div>
<h3 id="trident-state">Trident State</h3>
<p>We support trident persistent state that can be used with trident topologies. To create a RocketMQ persistent trident state you need to initialize it with the TupleToMessageMapper, TopicSelector, Properties instances. See the example below:</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"> <span class="n">TupleToMessageMapper</span> <span class="n">mapper</span> <span class="o">=</span> <span class="k">new</span> <span class="n">FieldNameBasedTupleToMessageMapper</span><span class="o">(</span><span class="s">"word"</span><span class="o">,</span> <span class="s">"count"</span><span class="o">);</span>
<span class="n">TopicSelector</span> <span class="n">selector</span> <span class="o">=</span> <span class="k">new</span> <span class="n">DefaultTopicSelector</span><span class="o">(</span><span class="n">topic</span><span class="o">);</span>
<span class="n">Properties</span> <span class="n">properties</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
<span class="n">properties</span><span class="o">.</span><span class="na">setProperty</span><span class="o">(</span><span class="n">RocketMqConfig</span><span class="o">.</span><span class="na">NAME_SERVER_ADDR</span><span class="o">,</span> <span class="n">nameserverAddr</span><span class="o">);</span>
<span class="n">RocketMqState</span><span class="o">.</span><span class="na">Options</span> <span class="n">options</span> <span class="o">=</span> <span class="k">new</span> <span class="n">RocketMqState</span><span class="o">.</span><span class="na">Options</span><span class="o">()</span>
<span class="o">.</span><span class="na">withMapper</span><span class="o">(</span><span class="n">mapper</span><span class="o">)</span>
<span class="o">.</span><span class="na">withSelector</span><span class="o">(</span><span class="n">selector</span><span class="o">)</span>
<span class="o">.</span><span class="na">withProperties</span><span class="o">(</span><span class="n">properties</span><span class="o">);</span>
<span class="n">StateFactory</span> <span class="n">factory</span> <span class="o">=</span> <span class="k">new</span> <span class="n">RocketMqStateFactory</span><span class="o">(</span><span class="n">options</span><span class="o">);</span>
<span class="n">TridentTopology</span> <span class="n">topology</span> <span class="o">=</span> <span class="k">new</span> <span class="n">TridentTopology</span><span class="o">();</span>
<span class="n">Stream</span> <span class="n">stream</span> <span class="o">=</span> <span class="n">topology</span><span class="o">.</span><span class="na">newStream</span><span class="o">(</span><span class="s">"spout1"</span><span class="o">,</span> <span class="n">spout</span><span class="o">);</span>
<span class="n">stream</span><span class="o">.</span><span class="na">partitionPersist</span><span class="o">(</span><span class="n">factory</span><span class="o">,</span> <span class="n">fields</span><span class="o">,</span>
<span class="k">new</span> <span class="nf">RocketMqStateUpdater</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">());</span>
</code></pre></div></div>
</div>
</div>
</div>
<footer>
<div class="container-fluid">
<div class="row">
<div class="col-md-3">
<div class="footer-widget">
<h5>Meetups</h5>
<ul class="latest-news">
<li><a href="http://www.meetup.com/Apache-Storm-Apache-Kafka/">Apache Storm & Apache Kafka</a> <span class="small">(Sunnyvale, CA)</span></li>
<li><a href="http://www.meetup.com/Apache-Storm-Kafka-Users/">Apache Storm & Kafka Users</a> <span class="small">(Seattle, WA)</span></li>
<li><a href="http://www.meetup.com/New-York-City-Storm-User-Group/">NYC Storm User Group</a> <span class="small">(New York, NY)</span></li>
<li><a href="http://www.meetup.com/Bay-Area-Stream-Processing">Bay Area Stream Processing</a> <span class="small">(Emeryville, CA)</span></li>
<li><a href="http://www.meetup.com/Boston-Storm-Users/">Boston Realtime Data</a> <span class="small">(Boston, MA)</span></li>
<li><a href="http://www.meetup.com/storm-london">London Storm User Group</a> <span class="small">(London, UK)</span></li>
<!-- <li><a href="http://www.meetup.com/Apache-Storm-Kafka-Users/">Seatle, WA</a> <span class="small">(27 Jun 2015)</span></li> -->
</ul>
</div>
</div>
<div class="col-md-3">
<div class="footer-widget">
<h5>About Apache Storm</h5>
<p>Apache Storm integrates with any queueing system and any database system. Apache Storm's spout abstraction makes it easy to integrate a new queuing system. Likewise, integrating Apache Storm with database systems is easy.</p>
</div>
</div>
<div class="col-md-3">
<div class="footer-widget">
<h5>First Look</h5>
<ul class="footer-list">
<li><a href="/releases/current/Rationale.html">Rationale</a></li>
<li><a href="/releases/current/Tutorial.html">Tutorial</a></li>
<li><a href="/releases/current/Setting-up-development-environment.html">Setting up development environment</a></li>
<li><a href="/releases/current/Creating-a-new-Storm-project.html">Creating a new Apache Storm project</a></li>
</ul>
</div>
</div>
<div class="col-md-3">
<div class="footer-widget">
<h5>Documentation</h5>
<ul class="footer-list">
<li><a href="/releases/current/index.html">Index</a></li>
<li><a href="/releases/current/javadocs/index.html">Javadoc</a></li>
<li><a href="/releases/current/FAQ.html">FAQ</a></li>
</ul>
</div>
</div>
</div>
<hr/>
<div class="row">
<div class="col-md-12">
<p align="center">Copyright © 2019 <a href="http://www.apache.org">Apache Software Foundation</a>. All Rights Reserved.
<br>Apache Storm, Apache, the Apache feather logo, and the Apache Storm project logos are trademarks of The Apache Software Foundation.
<br>All other marks mentioned may be trademarks or registered trademarks of their respective owners.</p>
</div>
</div>
</div>
</footer>
<!--Footer End-->
<!-- Scroll to top -->
<span class="totop"><a href="#"><i class="fa fa-angle-up"></i></a></span>
</body>
</html>