1"use strict";(self.webpackChunkrabbitmq_website=self.webpackChunkrabbitmq_website||[]).push([["17244"],{28662(e,n,s){s.r(n),s.d(n,{metadata:()=>i,default:()=>d,frontMatter:()=>o,contentTitle:()=>l,toc:()=>c,assets:()=>h});var i=JSON.parse('{"id":"tutorial-seven-java","title":"RabbitMQ tutorial - Reliable Publishing with Publisher Confirms","description":"\x3c!--","source":"@site/tutorials/tutorial-seven-java.md","sourceDirName":".","slug":"/tutorial-seven-java","permalink":"/tutorials/tutorial-seven-java","draft":false,"unlisted":false,"editUrl":"https://github.com/rabbitmq/rabbitmq-website/tree/main/tutorials/tutorial-seven-java.md","tags":[],"version":"current","frontMatter":{"title":"RabbitMQ tutorial - Reliable Publishing with Publisher Confirms"},"sidebar":"tutorialsSidebar","previous":{"title":"RPC","permalink":"/tutorials/tutorial-six-java"},"next":{"title":"Hello World","permalink":"/tutorials/tutorial-one-java-amqp10"}}'),a=s(74848),t=s(28453),r=s(11527);let o={title:"RabbitMQ tutorial - Reliable Publishing with Publisher Confirms"},l="RabbitMQ tutorial - Reliable Publishing with Publisher Confirms",h={},c=[{value:"Publisher Confirms",id:"publisher-confirms",level:2},...r.RM,{value:"(using the Java client)",id:"using-the-java-client",level:3},{value:"Overview",id:"overview",level:3},{value:"Enabling Publisher Confirms on a Channel",id:"enabling-publisher-confirms-on-a-channel",level:3},{value:"Strategy #1: Publishing Messages Individually",id:"strategy-1-publishing-messages-individually",level:3},{value:"Are Publisher Confirms Asynchronous?",id:"are-publisher-confirms-asynchronous",level:4},{value:"Strategy #2: Publishing Messages in Batches",id:"strategy-2-publishing-messages-in-batches",level:3},{value:"Strategy #3: Handling Publisher Confirms Asynchronously",id:"strategy-3-handling-publisher-confirms-asynchronously",level:3},{value:"How to Track Outstanding Confirms?",id:"how-to-track-outstanding-confirms",level:4},{value:"Re-publishing nack-ed Messages?",id:"re-publishing-nack-ed-messages",level:4},{value:"Summary",id:"summary",level:3},{value:"Putting It All Together",id:"putting-it-all-together",level:2}];function u(e){let n={a:"a",blockquote:"blockquote",code:"code",h1:"h1",h2:"h2",h3:"h3",h4:"h4",header:"header",li:"li",p:"p",pre:"pre",strong:"strong",ul:"ul",...(0,t.R)(),...e.components};return(0,a.jsxs)(a.Fragment,{children:[(0,a.jsx)(n.header,{children:(0,a.jsx)(n.h1,{id:"rabbitmq-tutorial---reliable-publishing-with-publisher-confirms",children:"RabbitMQ tutorial - Reliable Publishing with Publisher Confirms"})}),"\n",(0,a.jsx)(n.h2,{id:"publisher-confirms",children:"Publisher Confirms"}),"\n",(0,a.jsx)(r.Ay,{}),"\n",(0,a.jsxs)(n.p,{children:[(0,a.jsx)(n.a,{href:"/docs/confirms#publisher-confirms",children:"Publisher confirms"}),"\nare a RabbitMQ extension to implement reliable\npublishing. When publisher confirms are enabled on a channel,\nmessages the client publishes are confirmed asynchronously\nby the broker, meaning they have been taken care of on the server\nside."]}),"\n",(0,a.jsx)(n.h3,{id:"using-the-java-client",children:"(using the Java client)"}),"\n",(0,a.jsx)(n.h3,{id:"overview",children:"Overview"}),"\n",(0,a.jsx)(n.p,{children:"In this tutorial we're going to use publisher confirms to make\nsure published messages have safely reached the broker. We will\ncover several strategies for using publisher confirms and explain\ntheir pros and cons."}),"\n",(0,a.jsx)(n.h3,{id:"enabling-publisher-confirms-on-a-channel",children:"Enabling Publisher Confirms on a Channel"}),"\n",(0,a.jsxs)(n.p,{children:["Publisher confirms are a RabbitMQ extension to the AMQP 0.9.1 protocol,\nso they are not enabled by default. Publisher confirms are\nenabled at the channel level with the ",(0,a.jsx)(n.code,{children:"confirmSelect"})," method:"]}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-java",children:"Channel channel = connection.createChannel();\nchannel.confirmSelect();\n"})}),"\n",(0,a.jsx)(n.p,{children:"This method must be called on every channel that will use publisher confirms,\nand only once per channel, not for every published message."}),"\n",(0,a.jsx)(n.h3,{id:"strategy-1-publishing-messages-individually",children:"Strategy #1: Publishing Messages Individually"}),"\n",(0,a.jsx)(n.p,{children:"Let's start with the simplest approach to publishing with confirms,\nthat is, publishing a message and waiting synchronously for its confirmation:"}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-java",children:"while (thereAreMessagesToPublish()) {\n b
1yte[] body = ...;\n BasicProperties properties = ...;\n channel.basicPublish(exchange, queue, properties, body);\n // uses a 5 second timeout\n channel.waitForConfirmsOrDie(5_000);\n}\n"})}),"\n",(0,a.jsxs)(n.p,{children:["In the previous example we publish a message as usual and wait for its\nconfirmation with the ",(0,a.jsx)(n.code,{children:"Channel#waitForConfirmsOrDie(long)"})," method.\nThe method returns as soon as the message has been confirmed. If the\nmessage is not confirmed within the timeout or if it is nack-ed (meaning\nthe broker could not take care of it for some reason), the method will\nthrow an exception. Handling the exception usually means logging an\nerror message and/or retrying to send the message."]}),"\n",(0,a.jsx)(n.p,{children:"Different client libraries handle synchronous publisher confirms differently,\nso carefully read the documentation of the client you are using."}),"\n",(0,a.jsxs)(n.p,{children:["This technique is straightforward but has a major drawback:\nit ",(0,a.jsx)(n.strong,{children:"significantly slows down publishing"}),", as the confirmation of a message blocks the publishing\nof all subsequent messages. This approach won't deliver more than a few hundred\npublished messages per second. Nevertheless, this can be good enough for some\napplications."]}),"\n",(0,a.jsxs)(n.blockquote,{children:["\n",(0,a.jsx)(n.h4,{id:"are-publisher-confirms-asynchronous",children:"Are Publisher Confirms Asynchronous?"}),"\n",(0,a.jsxs)(n.p,{children:["We mentioned at the beginning that the broker confirms published\nmessages asynchronously but in the first example the code waits\nsynchronously until the message is confirmed. The client actually\nreceives confirms asynchronously and unblocks the call to ",(0,a.jsx)(n.code,{children:"waitForConfirmsOrDie"}),"\naccordingly. Think of ",(0,a.jsx)(n.code,{children:"waitForConfirmsOrDie"})," as a synchronous helper\nwhich relies on asynchronous notifications under the hood."]}),"\n"]}),"\n",(0,a.jsx)(n.h3,{id:"strategy-2-publishing-messages-in-batches",children:"Strategy #2: Publishing Messages in Batches"}),"\n",(0,a.jsx)(n.p,{children:"To improve upon our previous example, we can publish a batch\nof messages and wait for this whole batch to be confirmed.\nThe following example uses a batch of 100:"}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-java",children:"int batchSize = 100;\nint outstandingMessageCount = 0;\nwhile (thereAreMessagesToPublish()) {\n byte[] body = ...;\n BasicProperties properties = ...;\n channel.basicPublish(exchange, queue, properties, body);\n outstandingMessageCount++;\n if (outstandingMessageCount == batchSize) {\n channel.waitForConfirmsOrDie(5_000);\n outstandingMessageCount = 0;\n }\n}\nif (outstandingMessageCount > 0) {\n channel.waitForConfirmsOrDie(5_000);\n}\n"})}),"\n",(0,a.jsx)(n.p,{children:"Waiting for a whole batch to be confirmed drastically improves throughput over\nwaiting for each message individually (up to 20-30 times with a remote RabbitMQ node).\nOne drawback is that we do not know exactly what went wrong in case of failure,\nso we may have to keep a whole batch in memory to log something meaningful or\nto re-publish the messages. And this solution is still synchronous, so it\nblocks the publishing of messages."}),"\n",(0,a.jsx)(n.h3,{id:"strategy-3-handling-publisher-confirms-asynchronously",children:"Strategy #3: Handling Publisher Confirms Asynchronously"}),"\n",(0,a.jsx)(n.p,{children:"The broker confirms published messages asynchronously. To be notified of these\nconfirms, register a callback on the client:"}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-java",children:"Channel channel = connection.createChannel();\nchannel.confirmSelect();\nchannel.addConfirmListener((sequenceNumber, multiple) -> {\n // code when message is confirmed\n}, (sequenceNumber, multiple) -> {\n // code when message is nack-ed\n});\n"})}),"\n",(0,a.jsx)(n.p,{children:"There are 2 callbacks: one for confirmed messages and one for nack-ed messages\n(messages that can be considered lost by the broker). Each callback has\n2 parameters:"}),"\n",(0,a.jsxs)(n.ul,{children:["\n",(0,a.jsx)(n.li,{children:"sequence number: a number that identifies the confirmed\nor nack-ed message. We will see shortly how to correlate it with the published message."}),"\n",(0,a.jsx)(n.li,{children:"multiple: this is a boolean value. If false, only one message is confirmed/nack-ed;\nif true, all messages with a lower or equal sequence number are confirmed/nack-ed."}),"\n"]}),"\n",(0,a.jsxs)(n.p,{children:["The sequence number can be obtained with ",(0,a.jsx)(n.code,{children:"Channel#getNextPublishSeqNo()"}),"\nbefore publishing:"]}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-java",children:"int sequenceNumber = channel.getNextPublishSeqNo());\nch.basicPublish(exchange, queue, properties, body);\n"})}),"\n",(0,a.jsx)(n.p,{children:"A simple way to correlate messages with sequence numbers is to use a map.\nLet's assume we want to publish strings because they are easy to turn into\nan array of bytes for publishing. Here is a code sample that uses a map to\ncorrelate the publishing sequence number with the string body of the message:"}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-java",children:'ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();\n// ... code for confirm callbacks will come later\nString body = "...";\noutstandingConfirms.put(channel.getNextPublishSeqNo(), body);\nchannel.basicPublish(exchange, queue, properties, body.getBytes());\n'})}),"\n",(0,a.jsx)(n.p,{children:"The publishing code now tracks outbound messages with a map. We need\nto clean this map when confirms arrive and do something like logging a warning\nwhen messages are nack-ed:"}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-java",children:'ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();\nConfirmCallback cleanOutstandingConfirms = (sequenceNumber, multiple) -> {\n if (multiple) {\n ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(\n sequenceNumber, true\n );\n confirmed.clear();\n } else {\n outstandingConfirms.remove(sequenceNumber);\n }\n};\n\nchannel.addConfirmListener(cleanOutstandingConfirms, (sequenceNumber, multiple) -> {\n String body = outstandingConfirms.get(sequenceNumber);\n System.err.format(\n "Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n",\n body, sequenceNumber, multiple\n );\n cleanOutstandingConfirms.handle(sequenceNumber, multiple);\n});\n// ... publishing code\n'})}),"\n",(0,a.jsxs)(n.p,{children:["The previous sample contains a callback that cleans the map when\nconfirms arrive. Note this callback handles both single and multiple\nconfirms, and is registered as the first argument of\n",(0,a.jsx)(n.code,{children:"Channel#addConfirmListener"}),". The callback for nack-ed messages\nretrieves the message body and issues a warning. It then re-uses the\nprevious callback to clean the map of outstanding confirms (whether\nmessages are confirmed or nack-ed, their corresponding entr
1ies in the map\nmust be removed)."]}),"\n",(0,a.jsxs)(n.blockquote,{children:["\n",(0,a.jsx)(n.h4,{id:"how-to-track-outstanding-confirms",children:"How to Track Outstanding Confirms?"}),"\n",(0,a.jsxs)(n.p,{children:["Our samples use a ",(0,a.jsx)(n.code,{children:"ConcurrentNavigableMap"})," to track outstanding confirms.\nThis data structure is convenient for several reasons. It makes it easy to\ncorrelate a sequence number with a message (whatever the message data is)\nand to remove entries up to a given sequence id (to handle multiple\nconfirms/nacks). Lastly, it supports concurrent access, because confirm\ncallbacks are called in a thread owned by the client library, which should\nbe kept separate from the publishing thread."]}),"\n",(0,a.jsx)(n.p,{children:"There are other ways to track outstanding confirms than with\na sophisticated map implementation, like using a simple concurrent hash map\nand a variable to track the lower bound of the publishing sequence, but\nthey are usually more involved and do not belong in a tutorial."}),"\n"]}),"\n",(0,a.jsx)(n.p,{children:"To sum up, handling publisher confirms asynchronously usually requires the\nfollowing steps:"}),"\n",(0,a.jsxs)(n.ul,{children:["\n",(0,a.jsx)(n.li,{children:"provide a way to correlate the publishing sequence number with a message."}),"\n",(0,a.jsx)(n.li,{children:"register a confirm listener on the channel to be notified when\npublisher acks/nacks arrive to perform the appropriate actions, like\nlogging or re-publishing a nack-ed message. The sequence-number-to-message\ncorrelation mechanism may also require some cleaning during this step."}),"\n",(0,a.jsx)(n.li,{children:"track the publishing sequence number before publishing a message."}),"\n"]}),"\n",(0,a.jsxs)(n.blockquote,{children:["\n",(0,a.jsx)(n.h4,{id:"re-publishing-nack-ed-messages",children:"Re-publishing nack-ed Messages?"}),"\n",(0,a.jsxs)(n.p,{children:["It can be tempting to re-publish a nack-ed message from the corresponding\ncallback but this should be avoided, as confirm callbacks are\ndispatched in an I/O thread where channels are not supposed\nto do operations. A better solution is to enqueue the message in an in-memory\nqueue which is polled by a publishing thread. A class like ",(0,a.jsx)(n.code,{children:"ConcurrentLinkedQueue"}),"\nwould be a good candidate to transmit messages between the confirm callbacks\nand a publishing thread."]}),"\n"]}),"\n",(0,a.jsx)(n.h3,{id:"summary",children:"Summary"}),"\n",(0,a.jsx)(n.p,{children:"Making sure published messages made it to the broker can be essential in some applications.\nPublisher confirms are a RabbitMQ feature that helps meet this requirement. Publisher\nconfirms are asynchronous in nature, but it is also possible to handle them synchronously.\nThere is no single right way to implement publisher confirms; the choice usually comes\ndown to the constraints of the application and the overall system. Typical techniques are:"}),"\n",(0,a.jsxs)(n.ul,{children:["\n",(0,a.jsx)(n.li,{children:"publishing messages individually, waiting for the confirmation synchronously: simple, but very\nlimited throughput."}),"\n",(0,a.jsx)(n.li,{children:"publishing messages in batch, waiting for the confirmation synchronously for a batch: simple, reasonable\nthroughput, but hard to reason about when something goes wrong."}),"\n",(0,a.jsx)(n.li,{children:"asynchronous handling: best performance and use of resources, good control in case of error, but\ncan be involved to implement correctly."}),"\n"]}),"\n",(0,a.jsx)(n.h2,{id:"putting-it-all-together",children:"Putting It All Together"}),"\n",(0,a.jsxs)(n.p,{children:["The ",(0,a.jsx)(n.a,{href:"https://github.com/rabbitmq/rabbitmq-tutorials/blob/main/java/PublisherConfirms.java",children:(0,a.jsx)(n.code,{children:"PublisherConfirms.java"})}),"\nclass contains code for the techniques we covered. We can compile it, execute it as-is and\nsee how each of them performs:"]}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-bash",children:"javac -cp $CP PublisherConfirms.java\njava -cp $CP PublisherConfirms\n"})}),"\n",(0,a.jsx)(n.p,{children:"The output will look like the following:"}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-bash",children:"Published 50,000 messages individually in 5,549 ms\nPublished 50,000 messages in batch in 2,331 ms\nPublished 50,000 messages and handled confirms asynchronously in 4,054 ms\n"})}),"\n",(0,a.jsx)(n.p,{children:"The output on your computer should look similar if the\nclient and the server sit on the same machine. Publishing messages individually\nperforms poorly, as expected, but the results for asynchronous handling\nare a bit disappointing compared to batch publishing."}),"\n",(0,a.jsxs)(n.p,{children:["Publisher confirms are very network-dependent, so we are better off\ntrying with a remote node, which is more realistic as clients\nand servers are usually not on the same machine in production.\n",(0,a.jsx)(n.code,{children:"PublisherConfirms.java"})," can easily be changed to use a non-local node:"]}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-bash",children:'static Connection createConnection() throws Exception {\n ConnectionFactory cf = new ConnectionFactory();\n cf.setHost("remote-host");\n cf.setUsername("remote-user");\n cf.setPassword("remote-password");\n return cf.newConnection();\n}\n'})}),"\n",(0,a.jsx)(n.p,{children:"Recompile the class, execute it again, and wait for the results:"}),"\n",(0,a.jsx)(n.pre,{children:(0,a.jsx)(n.code,{className:"language-bash",children:"Published 50,000 messages individually in 231,541 ms\nPublished 50,000 messages in batch in 7,232 ms\nPublished 50,000 messages and handled confirms asynchronously in 6,332 ms\n"})}),"\n",(0,a.jsx)(n.p,{children:"We see publishing individually now performs terribly. But\nwith the network between the client and the server, batch publishing and asynchronous handling\nnow perform similarly, with a small advantage for asynchronous handling of the publisher confirms."}),"\n",(0,a.jsx)(n.p,{children:"Remember that batch publishing is simple to implement, but does not make it easy to know\nwhich message(s) could not make it to the broker in case of negative publisher acknowledgment.\nHandling publisher confirms asynchronously is more involved to implement but provides\nbetter granularity and more control over how to react when messages are nack-ed."})]})}function d(e={}){let{wrapper:n}={...(0,t.R)(),...e.components};return n?(0,a.jsx)(n,{...e,children:(0,a.jsx)(u,{...e})}):u(e)}},11527(e,n,s){s.d(n,{Ay:()=>o,RM:()=>t});var i=s(74848),a=s(28453);let t=[{value:"Prerequisites",id:"prerequisites",level:3},{value:"Where to get help",id:"where-to-get-help",level:3}];function r(e){let n={a:"a",admonition:"admonition",code:"code",h3:"h3",p:"p",...(0,a.R)(),...e.components};return(0,i.jsxs)(n.admonition,{type:"info",children:[(0,i.jsx)(n.h3,{id:"prerequisites",children:"Prerequisites"}),(0,i.jsxs)(n.p,{children:["This tutorial assumes RabbitMQ is ",(0,i.jsx)(n.a,{href:"/docs/download",children:"installed"})," and running on\n",(0,i.jsx)(n.code,{children:"localhost"})," on the ",(0,i.jsx)(n.a,{href:"/docs/networking#ports",children:"standard port"})," (5672). In case you\nuse a different host, port or credentials, connections settings would require\nadjusting."]}),(0,i.jsx)(n.h3,{id:"where-to-get-help",children:"Where to get help"}),(0,i.jsxs)(n.p,{children:["If you're having trouble going through this tutorial you can contact us\nthrough ",(0,i.jsx)(n.a,{href:"https://github.com/rabbitmq/rabbitmq-server/discussions",children:"GitHub Discussions"})," or ",(0,i.jsx)(n.a,{href:"https://www.rabbitmq.com/discord",children:"RabbitMQ community Discord"}),"."]})]})}function o(e={}){let{wrapper:n}={...(0,a.R)(),...e.components};return n?(0,i.jsx)(n,{...e,children:(0,i.jsx)(r,{...e})}):r(e)}},28453(e,n,s){s.d(n,{R:()=>r,x:()=>o});var i=s(96540);let a={},t=i.createContext(a);function r(e){let n=i.useContext(t);return i.useMemo(function(){return"function"==typeof e?e(n):{...n,...e}},[n,e])}function o(e){let n;return n=e.disableParentContext?"function"==typeof e.components?e.components(a):e.components||a:r(e.components),i.createElement(t.Provider,{value:n},e.children)}}}]);
Line numbers count LF bytes from the start of the resource, as the search results do. Vendor segments are library code the classifier recognised; they are stored but not indexed. Bytes are shown as Latin1 characters, one per byte.