| <div role="region" aria-label="Skip to main content"><a class="skipToContent_fXgn" href="#__docusaurus_skipToContent_fallback">Skip to main content</a></div><nav aria-label="Main" class="navbar navbar--fixed-top"><div class="navbar__inner"><div class="navbar__items"><button aria-label="Toggle navigation bar" aria-expanded="false" class="navbar__toggle clean-btn" type="button"><svg width="30" height="30" viewBox="0 0 30 30" aria-hidden="true"><path stroke="currentColor" stroke-linecap="round" stroke-miterlimit="10" stroke-width="2" d="M4 7h22M4 15h22M4 23h22"></path></svg></button><a class="navbar__brand" href="/"><div class="navbar__logo"><img src="/img/Apache_RocketMQ_logo.svg.png" alt="My Site Logo" class="themedImage_ToTc themedImage--light_HNdA"><img src="/img/Apache_RocketMQ_logo.svg.png" alt="My Site Logo" class="themedImage_ToTc themedImage--dark_i4oU"></div><b class="navbar__title text--truncate">Apache RocketMQ</b></a></div><div class="navbar__items navbar__items--right"><a href="https://github.com/apache/rocketmq" target="_blank" rel="noopener noreferrer" class="navbar__item navbar__link">GitHub<svg width="13.5" height="13.5" aria-hidden="true" viewBox="0 0 24 24" class="iconExternalLink_nPIU"><path fill="currentColor" d="M21 13v10h-21v-19h12v2h-10v15h17v-8h2zm3-12h-10.988l4.035 4-6.977 7.07 2.828 2.828 6.977-7.07 4.125 4.172v-11z"></path></svg></a><div class="navbar__item dropdown dropdown--hoverable dropdown--right"><a href="#" aria-haspopup="true" aria-expanded="false" role="button" class="navbar__link"><svg viewBox="0 0 24 24" width="20" height="20" aria-hidden="true" class="iconLanguage_nlXk"><path fill="currentColor" d="M12.87 15.07l-2.54-2.51.03-.03c1.74-1.94 2.98-4.17 3.71-6.53H17V4h-7V2H8v2H1v1.99h11.17C11.5 7.92 10.44 9.75 9 11.35 8.07 10.32 7.3 9.19 6.69 8h-2c.73 1.63 1.73 3.17 2.98 4.56l-5.09 5.02L4 19l5-5 3.11 3.11.76-2.04zM18.5 10h-2L12 22h2l1.12-3h4.75L21 22h2l-4.5-12zm-2.62 7l1.62-4.33L19.12 17h-3.24z"></path></svg>English</a><ul class="dropdown__menu"><li><a href="/zh/docs/featureBehavior/06consumertype" target="_self" rel="noopener noreferrer" class="dropdown__link" lang="zh">简体中文</a></li><li><a href="/docs/featureBehavior/06consumertype" target="_self" rel="noopener noreferrer" class="dropdown__link dropdown__link--active" lang="en">English</a></li></ul></div><div class="navbar__item dropdown dropdown--hoverable dropdown--right"><a href="#" aria-haspopup="true" aria-expanded="false" role="button" class="navbar__link">Docs</a><ul class="dropdown__menu"><li><a aria-current="page" class="dropdown__link dropdown__link--active" href="/docs/">5.0</a></li><li><a class="dropdown__link" href="/docs/4.x/">4.x</a></li></ul></div><a class="navbar__item navbar__link" href="/download">Download</a><div class="navbar__item dropdown dropdown--hoverable dropdown--right"><a href="#" aria-haspopup="true" aria-expanded="false" role="button" class="navbar__link">Blog</a><ul class="dropdown__menu"><li><a class="dropdown__link" href="/blog">User Cases</a></li><li><a class="dropdown__link" href="/events">Activity</a></li><li><a class="dropdown__link" href="/release-notes">Change Log</a></li><li><a class="dropdown__link" href="/news">RocketMQ News</a></li></ul></div><div class="navbar__item dropdown dropdown--hoverable dropdown--right"><a href="#" aria-haspopup="true" aria-expanded="false" role="button" class="navbar__link">Community</a><ul class="dropdown__menu"><li><a class="dropdown__link" href="/contact">Join Community</a></li><li><a class="dropdown__link" href="/origin">Origin</a></li><li><a class="dropdown__link" href="/team">Teams</a></li><li><a class="dropdown__link" href="/docs/contributionGuide/01how-to-contribute">Contributions</a></li><li><a class="dropdown__link" href="/enterprise">Enterprises</a></li></ul></div><div class="toggle_vylO colorModeToggle_DEke"><button class="clean-btn toggleButton_gllP toggleButtonDisabled_aARS" type="button" disabled="" title="Switch between dark and light mode (currently light mode)" aria-label="Switch between dark and light mode (currently light mode)" aria-live="polite"><svg viewBox="0 0 24 24" width="24" height="24" class="lightToggleIcon_pyhR"><path fill="currentColor" d="M12,9c1.65,0,3,1.35,3,3s-1.35,3-3,3s-3-1.35-3-3S10.35,9,12,9 M12,7c-2.76,0-5,2.24-5,5s2.24,5,5,5s5-2.24,5-5 S14.76,7,12,7L12,7z M2,13l2,0c0.55,0,1-0.45,1-1s-0.45-1-1-1l-2,0c-0.55,0-1,0.45-1,1S1.45,13,2,13z M20,13l2,0c0.55,0,1-0.45,1-1 s-0.45-1-1-1l-2,0c-0.55,0-1,0.45-1,1S19.45,13,20,13z M11,2v2c0,0.55,0.45,1,1,1s1-0.45,1-1V2c0-0.55-0.45-1-1-1S11,1.45,11,2z M11,20v2c0,0.55,0.45,1,1,1s1-0.45,1-1v-2c0-0.55-0.45-1-1-1C11.45,19,11,19.45,11,20z M5.99,4.58c-0.39-0.39-1.03-0.39-1.41,0 c-0.39,0.39-0.39,1.03,0,1.41l1.06,1.06c0.39,0.39,1.03,0.39,1.41,0s0.39-1.03,0-1.41L5.99,4.58z M18.36,16.95 c-0.39-0.39-1.03-0.39-1.41,0c-0.39,0.39-0.39,1.03,0,1.41l1.06,1.06c0.39,0.39,1.03,0.39,1.41,0c0.39-0.39,0.39-1.03,0-1.41 L18.36,16.95z M19.42,5.99c0.39-0.39,0.39-1.03,0-1.41c-0.39-0.39-1.03-0.39-1.41,0l-1.06,1.06c-0.39,0.39-0.39,1.03,0,1.41 s1.03,0.39,1.41,0L19.42,5.99z M7.05,18.36c0.39-0.39,0.39-1.03,0-1.41c-0.39-0.39-1.03-0.39-1.41,0l-1.06,1.06 c-0.39,0.39-0.39,1.03,0,1.41s1.03,0.39,1.41,0L7.05,18.36z"></path></svg><svg viewBox="0 0 24 24" width="24" height="24" class="darkToggleIcon_wfgR"><path fill="currentColor" d="M9.37,5.51C9.19,6.15,9.1,6.82,9.1,7.5c0,4.08,3.32,7.4,7.4,7.4c0.68,0,1.35-0.09,1.99-0.27C17.45,17.19,14.93,19,12,19 c-3.86,0-7-3.14-7-7C5,9.07,6.81,6.55,9.37,5.51z M12,3c-4.97,0-9,4.03-9,9s4.03,9,9,9s9-4.03,9-9c0-0.46-0.04-0.92-0.1-1.36 c-0.98,1.37-2.58,2.26-4.4,2.26c-2.98,0-5.4-2.42-5.4-5.4c0-1.81,0.89-3.42,2.26-4.4C12.92,3.04,12.46,3,12,3L12,3z"></path></svg></button></div><div class="searchBox_ZlJk"><button type="button" class="DocSearch DocSearch-Button" aria-label="Search"><span class="DocSearch-Button-Container"><svg width="20" height="20" class="DocSearch-Search-Icon" viewBox="0 0 20 20" aria-hidden="true"><path d="M14.386 14.386l4.0877 4.0877-4.0877-4.0877c-2.9418 2.9419-7.7115 2.9419-10.6533 0-2.9419-2.9418-2.9419-7.7115 0-10.6533 2.9418-2.9419 7.7115-2.9419 10.6533 0 2.9419 2.9418 2.9419 7.7115 0 10.6533z" stroke="currentColor" fill="none" fill-rule="evenodd" stroke-linecap="round" stroke-linejoin="round"></path></svg><span class="DocSearch-Button-Placeholder">Search</span></span><span class="DocSearch-Button-Keys"></span></button></div></div></div><div role="presentation" class="navbar-sidebar__backdrop"></div></nav><div id="__docusaurus_skipToContent_fallback" class="main-wrapper mainWrapper_z2l0 docsWrapper_BCFX"><button aria-label="Scroll back to top" class="clean-btn theme-back-to-top-button backToTopButton_sjWU" type="button"></button><div class="docPage__5DB"><aside class="theme-doc-sidebar-container docSidebarContainer_b6E3"><div class="sidebarViewport_Xe31"><div class="sidebar_njMd"><nav aria-label="Docs sidebar" class="menu thin-scrollbar menu_SIkG"><ul class="theme-doc-sidebar-menu menu__list"><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/">Introduction</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/quickStart/01quickstart">Quick Start</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/domainModel/01main">Domain Model</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret menu__link--active" aria-expanded="true" href="/docs/featureBehavior/01normalmessage">Feature Behavior</a></div><ul style="display:block;overflow:visible;height:auto" class="menu__list"><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/01normalmessage">Normal Message</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/02delaymessage">Delay Message</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/03fifomessage">Ordered Message</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/04transactionmessage">Transaction Message</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/05sendretrypolicy">Sending Retry and Throttling Policy</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link menu__link--active" aria-current="page" tabindex="0" href="/docs/featureBehavior/06consumertype">Consumer Types</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/07messagefilter">Message Filtering</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/08consumerloadbalance">Consumer Load Balancing</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/09consumerprogress">Consumer Progress Management</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/10consumerretrypolicy">Consumption Retry</a></li><li class="theme-doc-sidebar-item-link theme-doc-sidebar-item-link-level-2 menu__list-item"><a class="menu__link" tabindex="0" href="/docs/featureBehavior/11messagestorepolicy">Message Storage and Cleanup</a></li></ul></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/deploymentOperations/01deploy">Deployment & Operations</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/observability/01metrics">Observability</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/sdk/01overview">Client SDK</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/bestPractice/01bestpractice">Best Practice</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/eventbridge/01RocketMQEventBridgeConcepts">RocketMQ EventBridge</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/mqtt/01RocketMQMQTTOverview">RocketMQ MQTT</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/connect/01RocketMQ Connect Overview">RocketMQ Connect</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/streams/01RocketMQ Streams Overview">RocketMQ Streams</a></div></li><li class="theme-doc-sidebar-item-category theme-doc-sidebar-item-category-level-1 menu__list-item menu__list-item--collapsed"><div class="menu__list-item-collapsible"><a class="menu__link menu__link--sublist menu__link--sublist-caret" aria-expanded="false" href="/docs/contributionGuide/01how-to-contribute">Contribution Guide</a></div></li></ul></nav></div></div></aside><main class="docMainContainer_gTbr"><div class="container padding-top--md padding-bottom--lg"><div class="row"><div class="col docItemCol_VOVn"><div class="docItemContainer_Djhp"><article><nav class="theme-doc-breadcrumbs breadcrumbsContainer_Z_bl" aria-label="Breadcrumbs"><ul class="breadcrumbs" itemscope="" itemtype="https://schema.org/BreadcrumbList"><li class="breadcrumbs__item"><a aria-label="Home page" class="breadcrumbs__link" href="/"><svg viewBox="0 0 24 24" class="breadcrumbHomeIcon_YNFT"><path d="M10 19v-5h4v5c0 .55.45 1 1 1h3c.55 0 1-.45 1-1v-7h1.7c.46 0 .68-.57.33-.87L12.67 3.6c-.38-.34-.96-.34-1.34 0l-8.36 7.53c-.34.3-.13.87.33.87H5v7c0 .55.45 1 1 1h3c.55 0 1-.45 1-1z" fill="currentColor"></path></svg></a></li><li class="breadcrumbs__item"><span class="breadcrumbs__link">Feature Behavior</span><meta itemprop="position" content="1"></li><li itemscope="" itemprop="itemListElement" itemtype="https://schema.org/ListItem" class="breadcrumbs__item breadcrumbs__item--active"><span class="breadcrumbs__link" itemprop="name">Consumer Types</span><meta itemprop="position" content="2"></li></ul></nav><span class="theme-doc-version-badge badge badge--secondary">Version: 5.0</span><div class="tocCollapsible_ETCw theme-doc-toc-mobile tocMobile_ITEo"><button type="button" class="clean-btn tocCollapsibleButton_TO0P">On this page</button></div><div class="theme-doc-markdown markdown"><h1>Consumer Types</h1><p>Apache RocketMQ supports the following types of consumers: PushConsumer, SimpleConsumer and PullConsumer. This topic describes the usage, working and retry mechanisms, and scenarios of the three consumer types.</p><h2 class="anchor anchorWithStickyNavbar_LWe7" id="background-information">Background information<a href="#background-information" class="hash-link" aria-label="Direct link to heading" title="Direct link to heading"></a></h2><p>Apache RocketMQ provides the PushConsumer, SimpleConsumer and PullConsumer consumer types. The three consumer types have different integration and control methods that you can use to meet messaging requirements in different business scenarios. The following factors can help you choose a suitable consumer type for your business scenarios:</p><ul><li><p>Concurrent consumption: How can consumers use the multithreading technique to implement concurrent message consumption for higher message processing efficiency?</p></li><li><p>Synchronous or asynchronous message processing: For different integration scenarios, consumers may need to asynchronously distribute messages that they received to the business logic system for processing. How do I implement asynchronous message processing?</p></li><li><p>Reliable message processing: How do consumers return response results when they process messages? How do I implement message retries when message errors occur to ensure reliable message processing?</p></li></ul><p>For answers to the preceding problems, see <a href="#section-r97-urp-who">PushConsumer</a> and <a href="#section-b6m-nr8-8ii">SimpleConsumer</a>.</p><h2 class="anchor anchorWithStickyNavbar_LWe7" id="feature-overview">Feature overview<a href="#feature-overview" class="hash-link" aria-label="Direct link to heading" title="Direct link to heading"></a></h2><p><img loading="lazy" alt="消息消费流程" src="/assets/images/consumerflow-eaa625a6a01a048a155a3809a603529a.png" width="777" height="606" class="img_ev3q"></p><p>The preceding figure shows that message consumption by consumers in Apache RocketMQ involves the following stages: receiving messages, processing messages, and committing the consumption status.</p><p>The three types of consumers are suitable for various message consumption scenarios by providing different implementation methods and API operations. The following table describes the differences between the three types of consumers.</p><div class="theme-admonition theme-admonition-info alert alert--info admonition_LlT9"><div class="admonitionHeading_tbUL"><span class="admonitionIcon_kALy"><svg viewBox="0 0 14 16"><path fill-rule="evenodd" d="M7 2.3c3.14 0 5.7 2.56 5.7 5.7s-2.56 5.7-5.7 5.7A5.71 5.71 0 0 1 1.3 8c0-3.14 2.56-5.7 5.7-5.7zM7 1C3.14 1 0 4.14 0 8s3.14 7 7 7 7-3.14 7-7-3.14-7-7-7zm1 3H6v5h2V4zm0 6H6v2h2v-2z"></path></svg></span>info</div><div class="admonitionContent_S0QG"><p>PullConsumer is only recommended for integration in a stream processing framework. PushConsumer and simpleConsumer could satisfy most scenarios.</p><p>You can switch between the PushConsumer and SimpleConsumer based on your business scenarios. When you switch to a different consumer type, the usage of existing resources and existing business processing tasks in Apache RocketMQ are not affected.</p></div></div><div class="theme-admonition theme-admonition-danger alert alert--danger admonition_LlT9"><div class="admonitionHeading_tbUL"><span class="admonitionIcon_kALy"><svg viewBox="0 0 12 16"><path fill-rule="evenodd" d="M5.05.31c.81 2.17.41 3.38-.52 4.31C3.55 5.67 1.98 6.45.9 7.98c-1.45 2.05-1.7 6.53 3.53 7.7-2.2-1.16-2.67-4.52-.3-6.61-.61 2.03.53 3.33 1.94 2.86 1.39-.47 2.3.53 2.27 1.67-.02.78-.31 1.44-1.13 1.81 3.42-.59 4.78-3.42 4.78-5.56 0-2.84-2.53-3.22-1.25-5.61-1.52.13-2.03 1.13-1.89 2.75.09 1.08-1.02 1.8-1.86 1.33-.67-.41-.66-1.19-.06-1.78C8.18 5.31 8.68 2.45 5.05.32L5.03.3l.02.01z"></path></svg></span>danger</div><div class="admonitionContent_S0QG"><p>Mixing consumer type between pullConsumer and other consumer type in same consumerGroup is strictly prohibited.</p></div></div><table><thead><tr><th>Item</th><th>PushConsumer</th><th>SimpleConsumer</th><th>PullConsumer</th></tr></thead><tbody><tr><td>API operation call</td><td>The callback operation is called to return the consumption result by using a message listener. Consumers can process the consumption logic only within the scope of the message listener.</td><td>Business applications implement message processing and call the corresponding operation to return the consumption result.</td><td>Business applications implement message pulling and processing and call the corresponding operation to return the consumption result.</td></tr><tr><td>Consumption concurrency management</td><td>Apache RocketMQ SDKs are used to manage the number of concurrent threads for message consumption.</td><td>The number of concurrent threads that are used for message consumption is based on the consumption logic of individual business applications.</td><td>The number of concurrent threads that are used for message consumption is based on the consumption logic of individual business applications.</td></tr><tr><td>LoadBalance mechanism</td><td>Message-based load balancing in 5.0 version, Queue-based load balancing in earlier version.</td><td>Message-based load balancing.</td><td>Queue-based load balancing.</td></tr><tr><td>API flexibility</td><td>The API operations are encapsulated and provide poor flexibility.</td><td>The atomic operations provide great flexibility.</td><td>The atomic operations provide great flexibility.</td></tr><tr><td>Scenarios</td><td>This consumer type is suitable for development scenarios that do not require a custom process.</td><td>This consumer type is suitable for development scenarios that require custom processes.</td><td>It is recommended to be integrated only in the stream processing framework scenario</td></tr></tbody></table><h2 class="anchor anchorWithStickyNavbar_LWe7" id="pushconsumer">PushConsumer<a href="#pushconsumer" class="hash-link" aria-label="Direct link to heading" title="Direct link to heading"></a></h2><p>PushConsumer is a consumer type that provides a high degree of encapsulation. Message consumption and consumption result submission are processed by using only the message listener. The message acquisition, consumption status submission, and consumption retries are completed by using Apache RocketMQ client SDKs.</p><p><strong>Usage</strong></p><p>PushConsumer is used in a fixed manner. A message listener is registered with a PushConsumer consumer when the consumer is initialized, and message processing logic is implemented in the message listener. Message acquisition, listener call triggering, and message retries are processed by using Apache RocketMQ SDKs.</p><p>Sample code:</p><div class="language-java codeBlockContainer_Ckt0 theme-code-block" style="--prism-color:#393A34;--prism-background-color:#f6f8fa"><div class="codeBlockContent_biex"><pre tabindex="0" class="prism-code language-java codeBlock_bY9V thin-scrollbar"><code class="codeBlockLines_e6Vv"><span class="token-line" style="color:#393A34"><span class="token plain">// Message consumption example: Use a PushConsumer consumer to consume messages. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">ClientServiceProvider provider = ClientServiceProvider.loadService();</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">String topic = "YourTopic";</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">FilterExpression filterExpression = new FilterExpression("YourFilterTag", FilterExpressionType.TAG);</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">PushConsumer pushConsumer = provider.newPushConsumerBuilder()</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // Configure consumer group. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .setConsumerGroup("YourConsumerGroup")</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // Specify the access point. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build())</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // Specify the pre-bound subscriptions. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // Set the message listener. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .setMessageListener(new MessageListener() {</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> @Override</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> public ConsumeResult consume(MessageView messageView) {</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // Consume the messages and return the consumption result. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> return ConsumeResult.SUCCESS;</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> }</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> })</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .build(); </span><br></span></code></pre><div class="buttonGroup__atx"><button type="button" aria-label="Copy code to clipboard" title="Copy" class="clean-btn"><span class="copyButtonIcons_eSgA" aria-hidden="true"><svg viewBox="0 0 24 24" class="copyButtonIcon_y97N"><path fill="currentColor" d="M19,21H8V7H19M19,5H8A2,2 0 0,0 6,7V21A2,2 0 0,0 8,23H19A2,2 0 0,0 21,21V7A2,2 0 0,0 19,5M16,1H4A2,2 0 0,0 2,3V17H4V3H16V1Z"></path></svg><svg viewBox="0 0 24 24" class="copyButtonSuccessIcon_LjdS"><path fill="currentColor" d="M21,7L9,19L3.5,13.5L4.91,12.09L9,16.17L19.59,5.59L21,7Z"></path></svg></span></button></div></div></div><p>The message listener for a PushConsumer consumer returns one of the following results:</p><ul><li><p>Consumption success: For example, when you use Apache RocketMQ SDK for Java and a message is consumed, <code>ConsumeResult.SUCCESS</code> is returned. The server updates the consumption progress based on the consumption result.</p></li><li><p>Consumption failure: For example, when you use Apache RocketMQ SDK for Java and a message fails to be consumed, <code>ConsumeResult.FAILURE</code> is returned. Whether Apache RocketMQ retries to consume the message depends on the consumption retry logic.</p></li><li><p>Unexpected failure: For example, if an unexpected exception is thrown, the message fails to be consumed. Whether Apache RocketMQ retries to consume the message depends on the consumption retry logic.</p></li></ul><p>If an unexpected error in the message processing logic constantly prevents a message from being consumed by a PushConsumer consumer, the SDK considers that the consumption has timed out and forcibly commits a consumption failure result. Then, the message is processed based on the consumption retry logic. For more information about consumption timeouts, see <a href="/docs/featureBehavior/10consumerretrypolicy">Retry policy for push consumers</a>.</p><div class="theme-admonition theme-admonition-info alert alert--info admonition_LlT9"><div class="admonitionHeading_tbUL"><span class="admonitionIcon_kALy"><svg viewBox="0 0 14 16"><path fill-rule="evenodd" d="M7 2.3c3.14 0 5.7 2.56 5.7 5.7s-2.56 5.7-5.7 5.7A5.71 5.71 0 0 1 1.3 8c0-3.14 2.56-5.7 5.7-5.7zM7 1C3.14 1 0 4.14 0 8s3.14 7 7 7 7-3.14 7-7-3.14-7-7-7zm1 3H6v5h2V4zm0 6H6v2h2v-2z"></path></svg></span>info</div><div class="admonitionContent_S0QG"><p>When a consumption timeout occurs, the SDK commits the consumption failure result. However, the current consumption thread may fail to respond to the result and continues to process the message.:::</p><p><strong>Working mechanism</strong></p><p>For PushConsumer, real-time message processing is based on the typical Reactor thread model of the SDK. The SDK has a built-in long polling thread, which pulls messages and stores the messages to a queue. Then, the messages are delivered from the queue to individual message consumption threads. The message listener behaves based on the message consumption logic. The following figure shows the message consumption process of PushConsumer consumers. |
| :::</p><p>Scenarios</p><p>PushConsumer limits message processing to synchronous processing and restricts the timeout for processing each message. PushConsumer is suitable for the following scenarios:</p><ul><li><p>Predictable message processing duration: If the message processing duration is not limited, message retries are continuously triggered for messages that require a long processing duration to ensure message reliability. This causes a large number of repeated messages.</p></li><li><p>No asynchronization and no custom process: PushConsumer limits the thread model of the consumption logic to the Reactor thread model. The client SDK processes messages based on the maximum throughput. This model is easy to develop, but does not allow asynchronous or custom processes.</p></li></ul><h2 class="anchor anchorWithStickyNavbar_LWe7" id="simpleconsumer">SimpleConsumer<a href="#simpleconsumer" class="hash-link" aria-label="Direct link to heading" title="Direct link to heading"></a></h2><p>SimpleConsumer is a consumer type that supports atomic operations for message processing. Such type of consumers call operations to acquire messages, commit the consumption status, and perform message retries based on the business logic.</p><p><strong>Usage</strong></p><p>SimpleConsumer involves multiple API operations. The corresponding operations are called as needed to obtain and distribute messages to business threads for processing. Then, the commit operation is called to commit message processing results. Sample code:</p><div class="language-java codeBlockContainer_Ckt0 theme-code-block" style="--prism-color:#393A34;--prism-background-color:#f6f8fa"><div class="codeBlockContent_biex"><pre tabindex="0" class="prism-code language-java codeBlock_bY9V thin-scrollbar"><code class="codeBlockLines_e6Vv"><span class="token-line" style="color:#393A34"><span class="token plain">// Consumption example: When a SimpleConsumer consumer consumes normal messages, the consumer obtain messages and commit message consumption results. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">ClientServiceProvider provider = ClientServiceProvider.loadService();</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">String topic = "YourTopic";</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">FilterExpression filterExpression = new FilterExpression("YourFilterTag", FilterExpressionType.TAG);</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // Configure consumer group.</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .setConsumerGroup("YourConsumerGroup")</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // Specify the access point. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build())</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // Specify the pre-bound subscriptions. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // Specify the max await time when receive messages from the server.</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .setAwaitDuration(Duration.ofSeconds(1))</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> .build();</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">try {</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // A SimpleConsumer consumer must obtain and process messages. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> messageViewList.forEach(messageView -> {</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> System.out.println(messageView);</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // After consumption is complete, the consumer must invoke ACK to submit the consumption result. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> try {</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> simpleConsumer.ack(messageView);</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> } catch (ClientException e) {</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> logger.error("Failed to ack message, messageId={}", messageView.getMessageId(), e);</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> }</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> });</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">} catch (ClientException e) {</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> // If the pull fails due to system traffic throttling or other reasons, the consumer must re-initiate the request to obtain the message. </span><br></span><span class="token-line" style="color:#393A34"><span class="token plain"> logger.error("Failed to receive message", e);</span><br></span><span class="token-line" style="color:#393A34"><span class="token plain">}</span><br></span></code></pre><div class="buttonGroup__atx"><button type="button" aria-label="Copy code to clipboard" title="Copy" class="clean-btn"><span class="copyButtonIcons_eSgA" aria-hidden="true"><svg viewBox="0 0 24 24" class="copyButtonIcon_y97N"><path fill="currentColor" d="M19,21H8V7H19M19,5H8A2,2 0 0,0 6,7V21A2,2 0 0,0 8,23H19A2,2 0 0,0 21,21V7A2,2 0 0,0 19,5M16,1H4A2,2 0 0,0 2,3V17H4V3H16V1Z"></path></svg><svg viewBox="0 0 24 24" class="copyButtonSuccessIcon_LjdS"><path fill="currentColor" d="M21,7L9,19L3.5,13.5L4.91,12.09L9,16.17L19.59,5.59L21,7Z"></path></svg></span></button></div></div></div><p>The following table describes the API operations that are provided for SimpleConsumer.</p><table><thead><tr><th>Operation</th><th>Description</th><th>Modifiable parameter</th></tr></thead><tbody><tr><td><code>ReceiveMessage</code></td><td>Consumers can call this operation to obtain messages from the server. <strong>Note</strong> Because the server uses distributed storage, the server may return an empty result, although the requested message actually exists on the server. You can call the ReceiveMessage operation again or increase the concurrency value in the ReceiveMessage operation.</td><td><em> Batch pull size: the number of messages that are obtained at a time. A SimpleConsumer consumer can obtain multiple messages for batch consumption. </em> Message invisibility duration: the maximum processing duration of a message. This parameter controls the message retry interval if consumption fails. For more information, see <strong>Retry policy for SimpleConsumer</strong>. This parameter is required when you call the <code>ReceiveMessage</code> operation.</td></tr><tr><td><code>AckMessage</code></td><td>After a message is consumed by a consumer, the consumer calls this operation to return the consumption success result to the server.</td><td>None</td></tr><tr><td><code>ChangeInvisibleDuration</code></td><td>In consumption retry scenarios, consumers can call this operation to change the message processing duration to control the message retry interval.</td><td>Message invisibility duration: the maximum processing time of a message. You can call this operation to change the message invisibility duration that is specified in the <code>ReceiveMessage</code> operation. In most cases, this operation is used in scenarios in which you want to increase the message processing duration.</td></tr></tbody></table><p><strong>Retry for reliability</strong></p><p>When SimpleConsumer consumers consume messages, the communication between the client SDK and the Apache RocketMQ server is implemented by using the <code>ReceiveMessage</code> and <code>AckMessage</code> operations. When the client SDK successfully processes a message, the <code>AckMessage</code> operation is called. When the message fails to be processed, no ack message is returned to trigger the message retry mechanism after the specified message invisibility duration elapses. For more information, see <a href="/docs/featureBehavior/10consumerretrypolicy">Retry policy for simple consumers</a>.</p><p><strong>Ensured message order</strong></p><p>In Apache RocketMQ, a SimpleConsumer consumer obtains <a href="/docs/featureBehavior/03fifomessage">Fifo messages</a> in the order in which they are stored. If a message in a set of ordered messages is not completely processed, the next message in the set of order messages cannot be obtained.</p><p>Scenarios</p><p>SimpleConsumer provides atomic API operations to obtain messages and commit consumption results. Compared with PushConsumer, SimpleConsumer provides better flexibility. SimpleConsumer is suitable for the following scenarios:</p><ul><li><p>Uncontrollable message processing duration: If the message processing duration is inestimable, we recommend that you use SimpleConsumer to prevent messages from being processed for an excessively long period of time. You can specify an estimated message processing duration during message consumption. If an existing processing duration is not suitable for your business scenarios, you can call the corresponding API operation to change the message processing duration.</p></li><li><p>Asynchronous processing and batch consumption: SimpleConsumer does not involve complex thread encapsulation in the SDK. Business applications can use custom settings. This way, SimpleConsumer consumers can implement asynchronous distribution, batch consumption, and other custom scenarios.</p></li><li><p>Custom message consumption rate: When SimpleConsumer is used, business applications call the ReceiveMessage operation to obtain messages. You can adjust the frequency of obtaining message to control the message consumption rate.</p></li></ul><h2 class="anchor anchorWithStickyNavbar_LWe7" id="pullconsumer">PullConsumer<a href="#pullconsumer" class="hash-link" aria-label="Direct link to heading" title="Direct link to heading"></a></h2><p>To be continued.</p><h2 class="anchor anchorWithStickyNavbar_LWe7" id="usage-notes">Usage notes<a href="#usage-notes" class="hash-link" aria-label="Direct link to heading" title="Direct link to heading"></a></h2><p><strong>Specify a proper consumption duration limit for PushConsumer</strong></p><p>We recommend that you limit the message consumption duration for PushConsumer consumers to prevent a message from being processed for a long time. Long-time processing of a message can cause duplicate messages due to message processing timeouts and keep the next message continuously waiting from consumption. If messages are frequently processed for an excessively long period of time, we recommend that you use SimpleConsumer and specify a suitable message invisibility duration based on your business requirements.</p></div><footer class="theme-doc-footer docusaurus-mt-lg"><div class="theme-doc-footer-edit-meta-row row"><div class="col"><a href="https://github.com/apache/rocketmq-site/tree/new-official-website/versioned_docs/version-5.0/04-featureBehavior/06consumertype.md" target="_blank" rel="noreferrer noopener" class="theme-edit-this-page"><svg fill="currentColor" height="20" width="20" viewBox="0 0 40 40" class="iconEdit_Z9Sw" aria-hidden="true"><g><path d="m34.5 11.7l-3 3.1-6.3-6.3 3.1-3q0.5-0.5 1.2-0.5t1.1 0.5l3.9 3.9q0.5 0.4 0.5 1.1t-0.5 1.2z m-29.5 17.1l18.4-18.5 6.3 6.3-18.4 18.4h-6.3v-6.2z"></path></g></svg>Edit this page</a></div><div class="col lastUpdated_vwxv"></div></div></footer></article><nav class="pagination-nav docusaurus-mt-lg" aria-label="Docs pages navigation"><a class="pagination-nav__link pagination-nav__link--prev" href="/docs/featureBehavior/05sendretrypolicy"><div class="pagination-nav__sublabel">Previous</div><div class="pagination-nav__label">Sending Retry and Throttling Policy</div></a><a class="pagination-nav__link pagination-nav__link--next" href="/docs/featureBehavior/07messagefilter"><div class="pagination-nav__sublabel">Next</div><div class="pagination-nav__label">Message Filtering</div></a></nav></div></div><div class="col col--3"><div class="tableOfContents_bqdL thin-scrollbar theme-doc-toc-desktop"><ul class="table-of-contents table-of-contents__left-border"><li><a href="#background-information" class="table-of-contents__link toc-highlight">Background information</a></li><li><a href="#feature-overview" class="table-of-contents__link toc-highlight">Feature overview</a></li><li><a href="#pushconsumer" class="table-of-contents__link toc-highlight">PushConsumer</a></li><li><a href="#simpleconsumer" class="table-of-contents__link toc-highlight">SimpleConsumer</a></li><li><a href="#pullconsumer" class="table-of-contents__link toc-highlight">PullConsumer</a></li><li><a href="#usage-notes" class="table-of-contents__link toc-highlight">Usage notes</a></li></ul></div></div></div></div></main></div></div><footer class="footer footer--dark"><div class="container container-fluid"><div class="row footer__links"><div class="col footer__col"><div class="footer__title">Learn</div><ul class="footer__items clean-list"><li class="footer__item"><a class="footer__link-item" href="/docs/domainModel/01main">Introduction</a></li><li class="footer__item"><a class="footer__link-item" href="/docs/quickStart/01quickstart">Installation</a></li><li class="footer__item"><a class="footer__link-item" href="/version">Migration from 4.x to 5.0</a></li></ul></div><div class="col footer__col"><div class="footer__title">Community</div><ul class="footer__items clean-list"><li class="footer__item"><a href="https://twitter.com/ApacheRocketMQ" target="_blank" rel="noopener noreferrer" class="footer__link-item">Twitter<svg width="13.5" height="13.5" aria-hidden="true" viewBox="0 0 24 24" class="iconExternalLink_nPIU"><path fill="currentColor" d="M21 13v10h-21v-19h12v2h-10v15h17v-8h2zm3-12h-10.988l4.035 4-6.977 7.07 2.828 2.828 6.977-7.07 4.125 4.172v-11z"></path></svg></a></li><li class="footer__item"><a href="https://github.com/apache/rocketmq" target="_blank" rel="noopener noreferrer" class="footer__link-item">Github</a></li><li class="footer__item"><a href="https://github.com/apache/rocketmq" target="_blank" rel="noopener noreferrer" class="footer__link-item">Help</a></li></ul></div><div class="col footer__col"><div class="footer__title">More</div><ul class="footer__items clean-list"><li class="footer__item"><a class="footer__link-item" href="/blog">Blog</a></li><li class="footer__item"><a class="footer__link-item" href="/release-notes">Changelog</a></li><li class="footer__item"><a href="https://github.com/apache/rocketmq" target="_blank" rel="noopener noreferrer" class="footer__link-item">GitHub<svg width="13.5" height="13.5" aria-hidden="true" viewBox="0 0 24 24" class="iconExternalLink_nPIU"><path fill="currentColor" d="M21 13v10h-21v-19h12v2h-10v15h17v-8h2zm3-12h-10.988l4.035 4-6.977 7.07 2.828 2.828 6.977-7.07 4.125 4.172v-11z"></path></svg></a></li><li class="footer__item"><a href="https://twitter.com/docusaurus" target="_blank" rel="noopener noreferrer" class="footer__link-item">Twitter<svg width="13.5" height="13.5" aria-hidden="true" viewBox="0 0 24 24" class="iconExternalLink_nPIU"><path fill="currentColor" d="M21 13v10h-21v-19h12v2h-10v15h17v-8h2zm3-12h-10.988l4.035 4-6.977 7.07 2.828 2.828 6.977-7.07 4.125 4.172v-11z"></path></svg></a></li></ul></div><div class="col footer__col"><div class="footer__title">Legal</div><ul class="footer__items clean-list"><li class="footer__item"><a href="https://www.apache.org/licenses/" target="_blank" rel="noopener noreferrer" class="footer__link-item">Licenses<svg width="13.5" height="13.5" aria-hidden="true" viewBox="0 0 24 24" class="iconExternalLink_nPIU"><path fill="currentColor" d="M21 13v10h-21v-19h12v2h-10v15h17v-8h2zm3-12h-10.988l4.035 4-6.977 7.07 2.828 2.828 6.977-7.07 4.125 4.172v-11z"></path></svg></a></li><li class="footer__item"><a href="https://www.apache.org/security/" target="_blank" rel="noopener noreferrer" class="footer__link-item">Security<svg width="13.5" height="13.5" aria-hidden="true" viewBox="0 0 24 24" class="iconExternalLink_nPIU"><path fill="currentColor" d="M21 13v10h-21v-19h12v2h-10v15h17v-8h2zm3-12h-10.988l4.035 4-6.977 7.07 2.828 2.828 6.977-7.07 4.125 4.172v-11z"></path></svg></a></li><li class="footer__item"><a href="https://www.apache.org/foundation/thanks.html" target="_blank" rel="noopener noreferrer" class="footer__link-item">Thanks<svg width="13.5" height="13.5" aria-hidden="true" viewBox="0 0 24 24" class="iconExternalLink_nPIU"><path fill="currentColor" d="M21 13v10h-21v-19h12v2h-10v15h17v-8h2zm3-12h-10.988l4.035 4-6.977 7.07 2.828 2.828 6.977-7.07 4.125 4.172v-11z"></path></svg></a></li><li class="footer__item"><a href="https://www.apache.org/foundation/sponsorship.html" target="_blank" rel="noopener noreferrer" class="footer__link-item">Sponsorship<svg width="13.5" height="13.5" aria-hidden="true" viewBox="0 0 24 24" class="iconExternalLink_nPIU"><path fill="currentColor" d="M21 13v10h-21v-19h12v2h-10v15h17v-8h2zm3-12h-10.988l4.035 4-6.977 7.07 2.828 2.828 6.977-7.07 4.125 4.172v-11z"></path></svg></a></li></ul></div></div><div class="footer__bottom text--center"><div class="margin-bottom--sm"><a href="https://rocketmq.apache.org/" rel="noopener noreferrer" class="footerLogoLink_BH7S"><img src="/img/Apache_RocketMQ_logo.svg.png" alt="Meta Open Source Logo" class="themedImage_ToTc themedImage--light_HNdA footer__logo"><img src="/img/Apache_RocketMQ_logo.svg.png" alt="Meta Open Source Logo" class="themedImage_ToTc themedImage--dark_i4oU footer__logo"></a></div><div class="footer__copyright">Copyright © 2022 The Apache Software Foundation. Licensed under the Apache License, Version 2.0.</div></div></div></footer></div> |
