(window.webpackJsonp=window.webpackJsonp||[]).push([[56],{484:function(e,t,a){"use strict";a.r(t);varr=a(56),n=Object(r.a)({},(function(){vare=this,t=e.$createElement,a=e._self._c||t;returna("ContentSlotsDistributor",{attrs:{"slot-key":e.$parent.slotKey}},[a("h1",{attrs:{id:"messaging"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#messaging"}},[e._v("#")]),e._v(" Messaging")]),e._v(""),a("p",[e._v("The Spring Framework provides extensive support for integrating with messaging systems, from simplified use of the JMS API using "),a("code",[e._v("JmsTemplate")]),e._v(" to a complete infrastructure to receive messages asynchronously.\nSpring AMQP provides a similar feature set for the Advanced Message Queuing Protocol.\nSpring Boot also provides auto-configuration options for "),a("code",[e._v("RabbitTemplate")]),e._v(" and RabbitMQ.\nSpring WebSocket natively includes support for STOMP messaging, and Spring Boot has support for that through starters and a small amount of auto-configuration.\nSpring Boot also has support for Apache Kafka.")]),e._v(""),a("h2",{attrs:{id:"_1-jms"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_1-jms"}},[e._v("#")]),e._v(" 1. JMS")]),e._v(""),a("p",[e._v("The "),a("code",[e._v("javax.jms.ConnectionFactory")]),e._v(" interface provides a standard method of creating a "),a("code",[e._v("javax.jms.Connection")]),e._v(" for interacting with a JMS broker.\nAlthough Spring needs a "),a("code",[e._v("ConnectionFactory")]),e._v(" to work with JMS, you generally need not use it directly yourself and can instead rely on higher level messaging abstractions.\n(See the "),a("a",{attrs:{href:"https://docs.spring.io/spring-framework/docs/5.3.16/reference/html/integration.html#jms",target:"_blank",rel:"noopener noreferrer"}},[e._v("relevant section"),a("OutboundLink")],1),e._v(" of the Spring Framework reference documentation for details.)\nSpring Boot also auto-configures the necessary infrastructure to send and receive messages.")]),e._v(""),a("h3",{attrs:{id:"_1-1-activemq-support"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_1-1-activemq-support"}},[e._v("#")]),e._v(" 1.1. ActiveMQ Support")]),e._v(""),a("p",[e._v("When "),a("a",{attrs:{href:"https://activemq.apache.org/",target:"_blank",rel:"noopener noreferrer"}},[e._v("ActiveMQ"),a("OutboundLink")],1),e._v(" is available on the classpath, Spring Boot can also configure a "),a("code",[e._v("ConnectionFactory")]),e._v(".\nIf the broker is present, an embedded broker is automatically started and configured (provided no broker URL is specified through configuration and the embedded broker is not disabled in the configuration).")]),e._v(""),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("If you use "),a("code",[e._v("spring-boot-starter-activemq")]),e._v(", the necessary dependencies to connect or embed an ActiveMQ instance are provided, as is the Spring infrastructure to integrate with JMS.")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("ActiveMQ configuration is controlled by external configuration properties in "),a("code",[e._v("spring.activemq.*")]),e._v(".")]),e._v(""),a("p",[e._v("By default, ActiveMQ is auto-configured to use the "),a("a",{attrs:{href:"https://activemq.apache.org/vm-transport-reference.html",target:"_blank",rel:"noopener noreferrer"}},[e._v("VM transport"),a("OutboundLink")],1),e._v(", which starts a broker embedded in the same JVM instance.")]),e._v(""),a("p",[e._v("You can disable the embedded broker by configuring the "),a("code",[e._v("spring.activemq.in-memory")]),e._v(" property, as shown in the following example:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.activemq.in-memory=false\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring:\n activemq:\n in-memory: false\n")])])]),a("p",[e._v("The embedded broker will also be disabled if you configure the broker URL, as shown in the following example:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.activemq.broker-url=tcp://192.168.1.210:9876\nspring.activemq.user=admin\nspring.activemq.password=secret\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n activemq:\n broker-url: "tcp://192.168.1.210:9876"\n user: "admin"\n password: "secret"\n')])])]),a("p",[e._v("If you want to take full control over the embedded broker, see "),a("a",{attrs:{href:"https://activemq.apache.org/how-do-i-embed-a-broker-inside-a-connection.html",target:"_blank",rel:"noopener noreferrer"}},[e._v("the ActiveMQ documentation"),a("OutboundLink")],1),e._v(" for further information.")]),e._v(""),a("p",[e._v("By default, a "),a("code",[e._v("CachingConnectionFactory")]),e._v(" wraps the native "),a("code",[e._v("ConnectionFactory")]),e._v(" with sensible settings that you can control by external configuration properties in "),a("code",[e._v("spring.jms.*")]),e._v(":")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.jms.cache.session-cache-size=5\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring:\n jms:\n cache:\n session-cache-size: 5\n")])])]),a("p",[e._v("If you’d rather use native pooling, you can do so by adding a dependency to "),a("code",[e._v("org.messaginghub:pooled-jms")]),e._v(" and configuring the "),a("code",[e._v("JmsPoolConnectionFactory")]),e._v(" accordingly, as shown in the following example:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.activemq.pool.enabled=true\nspring.activemq.pool.max-connections=50\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring:\n activemq:\n pool:\n enabled: true\n max-connections: 50\n")])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("See "),a("a",{attrs:{href:"https://github.com/spring-projects/spring-boot/tree/v2.6.4/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/jms/activemq/ActiveMQProperties.java",target:"_blank",rel:"noopener noreferrer"}},[a("code",[e._v("ActiveMQProperties")]),a("OutboundLink")],1),e._v(" for more of the supported options."),a("br"),e._v("You can also register an arbitrary number of beans that implement "),a("code",[e._v("ActiveMQConnectionFactoryCustomizer")]),e._v(" for more advanced customizations.")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("By default, ActiveMQ creates a destination if it does not yet exist so that destinations are resolved against their provided names.")]),e._v(""),a("h3",{attrs:{id:"_1-2-activemq-artemis-support"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_1-2-activemq-artemis-support"}},[e._v("#")]),e._v(" 1.2. ActiveMQ Artemis Support")]),e._v(""),a("p",[e._v("Spring Boot can auto-configure a "),a("code",[e._v("ConnectionFactory")]),e._v(" when it detects that "),a("a",{attrs:{href:"https://activemq.apache.org/components/artemis/",target:"_blank",rel:"noopener noreferrer"}},[e._v("ActiveMQ Artemis"),a("OutboundLink")],1),e._v(" is available on the classpath.\nIf the broker is present, an embedded broker is automatically started and configured (unless the mode property has been explicitly set).\nThe supported modes are "),a("code",[e._v("embedded")]),e._v(" (to make explicit that an embedded broker is required and that an error should occur if the broker is not available on the classpath) and "),a("code",[e._v("native")]),e._v(" (to connect to a broker using the "),a("code",[e._v("netty")]),e._v(" transport protocol).\nWhen the latter is configured, Spring Boot configures a "),a("code",[e._v("ConnectionFactory")]),e._v(" that connects to a broker running on the local machine with the default settings.")]),e._v(""),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("If you use "),a("code",[e._v("spring-boot-starter-artemis")]),e._v(", the necessary dependencies to connect to an existing ActiveMQ Artemis instance are provided, as well as the Spring infrastructure to integrate with JMS."),a("br"),e._v("Adding "),a("code",[e._v("org.apache.activemq:artemis-jms-server")]),e._v(" to your application lets you use embedded mode.")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("ActiveMQ Artemis configuration is controlled by external configuration properties in "),a("code",[e._v("spring.artemis.*")]),e._v(".\nFor example, you might declare the following section in "),a("code",[e._v("application.properties")]),e._v(":")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.artemis.mode=native\nspring.artemis.broker-url=tcp://192.168.1.210:9876\nspring.artemis.user=admin\nspring.artemis.password=secret\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n artemis:\n mode: native\n broker-url: "tcp://192.168.1.210:9876"\n user: "admin"\n password: "secret"\n')])])]),a("p",[e._v("When embedding the broker, you can choose if you want to enable persistence and list the destinations that should be made available.\nThese can be specified as a comma-separated list to create them with the default options, or you can define bean(s) of type "),a("code",[e._v("org.apache.activemq.artemis.jms.server.config.JMSQueueConfiguration")]),e._v(" or "),a("code",[e._v("org.apache.activemq.artemis.jms.server.config.TopicConfiguration")]),e._v(", for advanced queue and topic configurations, respectively.")]),e._v(""),a("p",[e._v("By default, a "),a("code",[e._v("CachingConnectionFactory")]),e._v(" wraps the native "),a("code",[e._v("ConnectionFactory")]),e._v(" with sensible settings that you can control by external configuration properties in "),a("code",[e._v("spring.jms.*")]),e._v(":")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.jms.cache.session-cache-size=5\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring:\n jms:\n cache:\n session-cache-size: 5\n")])])]),a("p",[e._v("If you’d rather use native pooling, you can do so by adding a dependency to "),a("code",[e._v("org.messaginghub:pooled-jms")]),e._v(" and configuring the "),a("code",[e._v("JmsPoolConnectionFactory")]),e._v(" accordingly, as shown in the following example:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.artemis.pool.enabled=true\nspring.artemis.pool.max-connections=50\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring:\n artemis:\n pool:\n enabled: true\n max-connections: 50\n")])])]),a("p",[e._v("See "),a("a",{attrs:{href:"https://github.com/spring-projects/spring-boot/tree/v2.6.4/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/jms/artemis/ArtemisProperties.java",target:"_blank",rel:"noopener noreferrer"}},[a("code",[e._v("ArtemisProperties")]),a("OutboundLink")],1),e._v(" for more supported options.")]),e._v(""),a("p",[e._v("No JNDI lookup is involved, and destinations are resolved against their names, using either the "),a("code",[e._v("name")]),e._v(" attribute in the Artemis configuration or the names provided through configuration.")]),e._v(""),a("h3",{attrs:{id:"_1-3-using-a-jndi-connectionfactory"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_1-3-using-a-jndi-connectionfactory"}},[e._v("#")]),e._v(" 1.3. Using a JNDI ConnectionFactory")]),e._v(""),a("p",[e._v("If you are running your application in an application server, Spring Boot tries to locate a JMS "),a("code",[e._v("ConnectionFactory")]),e._v(" by using JNDI.\nBy default, the "),a("code",[e._v("java:/JmsXA")]),e._v(" and "),a("code",[e._v("java:/XAConnectionFactory")]),e._v(" location are checked.\nYou can use the "),a("code",[e._v("spring.jms.jndi-name")]),e._v(" property if you need to specify an alternative location, as shown in the following example:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.jms.jndi-name=java:/MyConnectionFactory\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n jms:\n jndi-name: "java:/MyConnectionFactory"\n')])])]),a("h3",{attrs:{id:"_1-4-sending-a-message"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_1-4-sending-a-message"}},[e._v("#")]),e._v(" 1.4. Sending a Message")]),e._v(""),a("p",[e._v("Spring’s "),a("code",[e._v("JmsTemplate")]),e._v(" is auto-configured, and you can autowire it directly into your own beans, as shown in the following example:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.springframework.jms.core.JmsTemplate;\nimport org.springframework.stereotype.Component;\n\n@Component\npublic class MyBean {\n\n private final JmsTemplate jmsTemplate;\n\n public MyBean(JmsTemplate jmsTemplate) {\n this.jmsTemplate = jmsTemplate;\n }\n\n // ...\n\n public void someMethod() {\n this.jmsTemplate.convertAndSend("hello");\n }\n\n}\n\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[a("a",{attrs:{href:"https://docs.spring.io/spring-framework/docs/5.3.16/javadoc-api/org/springframework/jms/core/JmsMessagingTemplate.html",target:"_blank",rel:"noopener noreferrer"}},[a("code",[e._v("JmsMessagingTemplate")]),a("OutboundLink")],1),e._v(" can be injected in a similar manner."),a("br"),e._v("If a "),a("code",[e._v("DestinationResolver")]),e._v(" or a "),a("code",[e._v("MessageConverter")]),e._v(" bean is defined, it is associated automatically to the auto-configured "),a("code",[e._v("JmsTemplate")]),e._v(".")])])]),e._v(""),a("tbody")]),e._v(""),a("h3",{attrs:{id:"_1-5-receiving-a-message"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_1-5-receiving-a-message"}},[e._v("#")]),e._v(" 1.5. Receiving a Message")]),e._v(""),a("p",[e._v("When the JMS infrastructure is present, any bean can be annotated with "),a("code",[e._v("@JmsListener")]),e._v(" to create a listener endpoint.\nIf no "),a("code",[e._v("JmsListenerContainerFactory")]),e._v(" has been defined, a default one is configured automatically.\nIf a "),a("code",[e._v("DestinationResolver")]),e._v(", a "),a("code",[e._v("MessageConverter")]),e._v(", or a "),a("code",[e._v("javax.jms.ExceptionListener")]),e._v(" beans are defined, they are associated automatically with the default factory.")]),e._v(""),a("p",[e._v("By default, the default factory is transactional.\nIf you run in an infrastructure where a "),a("code",[e._v("JtaTransactionManager")]),e._v(" is present, it is associated to the listener container by default.\nIf not, the "),a("code",[e._v("sessionTransacted")]),e._v(" flag is enabled.\nIn that latter scenario, you can associate your local data store transaction to the processing of an incoming message by adding "),a("code",[e._v("@Transactional")]),e._v(" on your listener method (or a delegate thereof).\nThis ensures that the incoming message is acknowledged, once the local transaction has completed.\nThis also includes sending response messages that have been performed on the same JMS session.")]),e._v(""),a("p",[e._v("The following component creates a listener endpoint on the "),a("code",[e._v("someQueue")]),e._v(" destination:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.springframework.jms.annotation.JmsListener;\nimport org.springframework.stereotype.Component;\n\n@Component\npublic class MyBean {\n\n @JmsListener(destination = "someQueue")\n public void processMessage(String content) {\n // ...\n }\n\n}\n\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("See "),a("a",{attrs:{href:"https://docs.spring.io/spring-framework/docs/5.3.16/javadoc-api/org/springframework/jms/annotation/EnableJms.html",target:"_blank",rel:"noopener noreferrer"}},[e._v("the Javadoc of "),a("code",[e._v("@EnableJms")]),a("OutboundLink")],1),e._v(" for more details.")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("If you need to create more "),a("code",[e._v("JmsListenerContainerFactory")]),e._v(" instances or if you want to override the default, Spring Boot provides a "),a("code",[e._v("DefaultJmsListenerContainerFactoryConfigurer")]),e._v(" that you can use to initialize a "),a("code",[e._v("DefaultJmsListenerContainerFactory")]),e._v(" with the same settings as the one that is auto-configured.")]),e._v(""),a("p",[e._v("For instance, the following example exposes another factory that uses a specific "),a("code",[e._v("MessageConverter")]),e._v(":")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("import javax.jms.ConnectionFactory;\n\nimport org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;\nimport org.springframework.context.annotation.Bean;\nimport org.springframework.context.annotation.Configuration;\nimport org.springframework.jms.config.DefaultJmsListenerContainerFactory;\n\n@Configuration(proxyBeanMethods = false)\npublic class MyJmsConfiguration {\n\n @Bean\n public DefaultJmsListenerContainerFactory myFactory(DefaultJmsListenerContainerFactoryConfigurer configurer) {\n DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();\n ConnectionFactory connectionFactory = getCustomConnectionFactory();\n configurer.configure(factory, connectionFactory);\n factory.setMessageConverter(new MyMessageConverter());\n return factory;\n }\n\n private ConnectionFactory getCustomConnectionFactory() {\n return ...\n }\n\n}\n\n")])])]),a("p",[e._v("Then you can use the factory in any "),a("code",[e._v("@JmsListener")]),e._v("-annotated method as follows:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.springframework.jms.annotation.JmsListener;\nimport org.springframework.stereotype.Component;\n\n@Component\npublic class MyBean {\n\n @JmsListener(destination = "someQueue", containerFactory = "myFactory")\n public void processMessage(String content) {\n // ...\n }\n\n}\n\n')])])]),a("h2",{attrs:{id:"_2-amqp"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_2-amqp"}},[e._v("#")]),e._v(" 2. AMQP")]),e._v(""),a("p",[e._v("The Advanced Message Queuing Protocol (AMQP) is a platform-neutral, wire-level protocol for message-oriented middleware.\nThe Spring AMQP project applies core Spring concepts to the development of AMQP-based messaging solutions.\nSpring Boot offers several conveniences for working with AMQP through RabbitMQ, including the "),a("code",[e._v("spring-boot-starter-amqp")]),e._v(" “Starter”.")]),e._v(""),a("h3",{attrs:{id:"_2-1-rabbitmq-support"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_2-1-rabbitmq-support"}},[e._v("#")]),e._v(" 2.1. RabbitMQ support")]),e._v(""),a("p",[a("a",{attrs:{href:"https://www.rabbitmq.com/",target:"_blank",rel:"noopener noreferrer"}},[e._v("RabbitMQ"),a("OutboundLink")],1),e._v(" is a lightweight, reliable, scalable, and portable message broker based on the AMQP protocol.\nSpring uses "),a("code",[e._v("RabbitMQ")]),e._v(" to communicate through the AMQP protocol.")]),e._v(""),a("p",[e._v("RabbitMQ configuration is controlled by external configuration properties in "),a("code",[e._v("spring.rabbitmq.*")]),e._v(".\nFor example, you might declare the following section in "),a("code",[e._v("application.properties")]),e._v(":")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.rabbitmq.host=localhost\nspring.rabbitmq.port=5672\nspring.rabbitmq.username=admin\nspring.rabbitmq.password=secret\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n rabbitmq:\n host: "localhost"\n port: 5672\n username: "admin"\n password: "secret"\n')])])]),a("p",[e._v("Alternatively, you could configure the same connection using the "),a("code",[e._v("addresses")]),e._v(" attribute:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.rabbitmq.addresses=amqp://admin:[email protected]\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n rabbitmq:\n addresses: "amqp://admin:[email protected]"\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("When specifying addresses that way, the "),a("code",[e._v("host")]),e._v(" and "),a("code",[e._v("port")]),e._v(" properties are ignored."),a("br"),e._v("If the address uses the "),a("code",[e._v("amqps")]),e._v(" protocol, SSL support is enabled automatically.")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("See "),a("a",{attrs:{href:"https://github.com/spring-projects/spring-boot/tree/v2.6.4/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java",target:"_blank",rel:"noopener noreferrer"}},[a("code",[e._v("RabbitProperties")]),a("OutboundLink")],1),e._v(" for more of the supported property-based configuration options.\nTo configure lower-level details of the RabbitMQ "),a("code",[e._v("ConnectionFactory")]),e._v(" that is used by Spring AMQP, define a "),a("code",[e._v("ConnectionFactoryCustomizer")]),e._v(" bean.")]),e._v(""),a("p",[e._v("If a "),a("code",[e._v("ConnectionNameStrategy")]),e._v(" bean exists in the context, it will be automatically used to name connections created by the auto-configured "),a("code",[e._v("CachingConnectionFactory")]),e._v(".")]),e._v(""),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("See "),a("a",{attrs:{href:"https://spring.io/blog/2010/06/14/understanding-amqp-the-protocol-used-by-rabbitmq/",target:"_blank",rel:"noopener noreferrer"}},[e._v("Understanding AMQP, the protocol used by RabbitMQ"),a("OutboundLink")],1),e._v(" for more details.")])])]),e._v(""),a("tbody")]),e._v(""),a("h3",{attrs:{id:"_2-2-sending-a-message"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_2-2-sending-a-message"}},[e._v("#")]),e._v(" 2.2. Sending a Message")]),e._v(""),a("p",[e._v("Spring’s "),a("code",[e._v("AmqpTemplate")]),e._v(" and "),a("code",[e._v("AmqpAdmin")]),e._v(" are auto-configured, and you can autowire them directly into your own beans, as shown in the following example:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.springframework.amqp.core.AmqpAdmin;\nimport org.springframework.amqp.core.AmqpTemplate;\nimport org.springframework.stereotype.Component;\n\n@Component\npublic class MyBean {\n\n private final AmqpAdmin amqpAdmin;\n\n private final AmqpTemplate amqpTemplate;\n\n public MyBean(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) {\n this.amqpAdmin = amqpAdmin;\n this.amqpTemplate = amqpTemplate;\n }\n\n // ...\n\n public void someMethod() {\n this.amqpAdmin.getQueueInfo("someQueue");\n }\n\n public void someOtherMethod() {\n this.amqpTemplate.convertAndSend("hello");\n }\n\n}\n\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[a("a",{attrs:{href:"https://docs.spring.io/spring-amqp/docs/2.4.2/api/org/springframework/amqp/rabbit/core/RabbitMessagingTemplate.html",target:"_blank",rel:"noopener noreferrer"}},[a("code",[e._v("RabbitMessagingTemplate")]),a("OutboundLink")],1),e._v(" can be injected in a similar manner."),a("br"),e._v("If a "),a("code",[e._v("MessageConverter")]),e._v(" bean is defined, it is associated automatically to the auto-configured "),a("code",[e._v("AmqpTemplate")]),e._v(".")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("If necessary, any "),a("code",[e._v("org.springframework.amqp.core.Queue")]),e._v(" that is defined as a bean is automatically used to declare a corresponding queue on the RabbitMQ instance.")]),e._v(""),a("p",[e._v("To retry operations, you can enable retries on the "),a("code",[e._v("AmqpTemplate")]),e._v(" (for example, in the event that the broker connection is lost):")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.rabbitmq.template.retry.enabled=true\nspring.rabbitmq.template.retry.initial-interval=2s\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n rabbitmq:\n template:\n retry:\n enabled: true\n initial-interval: "2s"\n')])])]),a("p",[e._v("Retries are disabled by default.\nYou can also customize the "),a("code",[e._v("RetryTemplate")]),e._v(" programmatically by declaring a "),a("code",[e._v("RabbitRetryTemplateCustomizer")]),e._v(" bean.")]),e._v(""),a("p",[e._v("If you need to create more "),a("code",[e._v("RabbitTemplate")]),e._v(" instances or if you want to override the default, Spring Boot provides a "),a("code",[e._v("RabbitTemplateConfigurer")]),e._v(" bean that you can use to initialize a "),a("code",[e._v("RabbitTemplate")]),e._v(" with the same settings as the factories used by the auto-configuration.")]),e._v(""),a("h3",{attrs:{id:"_2-3-receiving-a-message"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_2-3-receiving-a-message"}},[e._v("#")]),e._v(" 2.3. Receiving a Message")]),e._v(""),a("p",[e._v("When the Rabbit infrastructure is present, any bean can be annotated with "),a("code",[e._v("@RabbitListener")]),e._v(" to create a listener endpoint.\nIf no "),a("code",[e._v("RabbitListenerContainerFactory")]),e._v(" has been defined, a default "),a("code",[e._v("SimpleRabbitListenerContainerFactory")]),e._v(" is automatically configured and you can switch to a direct container using the "),a("code",[e._v("spring.rabbitmq.listener.type")]),e._v(" property.\nIf a "),a("code",[e._v("MessageConverter")]),e._v(" or a "),a("code",[e._v("MessageRecoverer")]),e._v(" bean is defined, it is automatically associated with the default factory.")]),e._v(""),a("p",[e._v("The following sample component creates a listener endpoint on the "),a("code",[e._v("someQueue")]),e._v(" queue:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.springframework.amqp.rabbit.annotation.RabbitListener;\nimport org.springframework.stereotype.Component;\n\n@Component\npublic class MyBean {\n\n @RabbitListener(queues = "someQueue")\n public void processMessage(String content) {\n // ...\n }\n\n}\n\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("See "),a("a",{attrs:{href:"https://docs.spring.io/spring-amqp/docs/2.4.2/api/org/springframework/amqp/rabbit/annotation/EnableRabbit.html",target:"_blank",rel:"noopener noreferrer"}},[e._v("the Javadoc of "),a("code",[e._v("@EnableRabbit")]),a("OutboundLink")],1),e._v(" for more details.")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("If you need to create more "),a("code",[e._v("RabbitListenerContainerFactory")]),e._v(" instances or if you want to override the default, Spring Boot provides a "),a("code",[e._v("SimpleRabbitListenerContainerFactoryConfigurer")]),e._v(" and a "),a("code",[e._v("DirectRabbitListenerContainerFactoryConfigurer")]),e._v(" that you can use to initialize a "),a("code",[e._v("SimpleRabbitListenerContainerFactory")]),e._v(" and a "),a("code",[e._v("DirectRabbitListenerContainerFactory")]),e._v(" with the same settings as the factories used by the auto-configuration.")]),e._v(""),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("It does not matter which container type you chose."),a("br"),e._v("Those two beans are exposed by the auto-configuration.")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("For instance, the following configuration class exposes another factory that uses a specific "),a("code",[e._v("MessageConverter")]),e._v(":")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;\nimport org.springframework.amqp.rabbit.connection.ConnectionFactory;\nimport org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;\nimport org.springframework.context.annotation.Bean;\nimport org.springframework.context.annotation.Configuration;\n\n@Configuration(proxyBeanMethods = false)\npublic class MyRabbitConfiguration {\n\n @Bean\n public SimpleRabbitListenerContainerFactory myFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer) {\n SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();\n ConnectionFactory connectionFactory = getCustomConnectionFactory();\n configurer.configure(factory, connectionFactory);\n factory.setMessageConverter(new MyMessageConverter());\n return factory;\n }\n\n private ConnectionFactory getCustomConnectionFactory() {\n return ...\n }\n\n}\n\n")])])]),a("p",[e._v("Then you can use the factory in any "),a("code",[e._v("@RabbitListener")]),e._v("-annotated method, as follows:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.springframework.amqp.rabbit.annotation.RabbitListener;\nimport org.springframework.stereotype.Component;\n\n@Component\npublic class MyBean {\n\n @RabbitListener(queues = "someQueue", containerFactory = "myFactory")\n public void processMessage(String content) {\n // ...\n }\n\n}\n\n')])])]),a("p",[e._v("You can enable retries to handle situations where your listener throws an exception.\nBy default, "),a("code",[e._v("RejectAndDontRequeueRecoverer")]),e._v(" is used, but you can define a "),a("code",[e._v("MessageRecoverer")]),e._v(" of your own.\nWhen retries are exhausted, the message is rejected and either dropped or routed to a dead-letter exchange if the broker is configured to do so.\nBy default, retries are disabled.\nYou can also customize the "),a("code",[e._v("RetryTemplate")]),e._v(" programmatically by declaring a "),a("code",[e._v("RabbitRetryTemplateCustomizer")]),e._v(" bean.")]),e._v(""),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("By default, if retries are disabled and the listener throws an exception, the delivery is retried indefinitely."),a("br"),e._v("You can modify this behavior in two ways: Set the "),a("code",[e._v("defaultRequeueRejected")]),e._v(" property to "),a("code",[e._v("false")]),e._v(" so that zero re-deliveries are attempted or throw an "),a("code",[e._v("AmqpRejectAndDontRequeueException")]),e._v(" to signal the message should be rejected."),a("br"),e._v("The latter is the mechanism used when retries are enabled and the maximum number of delivery attempts is reached.")])])]),e._v(""),a("tbody")]),e._v(""),a("h2",{attrs:{id:"_3-apache-kafka-support"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_3-apache-kafka-support"}},[e._v("#")]),e._v(" 3. Apache Kafka Support")]),e._v(""),a("p",[a("a",{attrs:{href:"https://kafka.apache.org/",target:"_blank",rel:"noopener noreferrer"}},[e._v("Apache Kafka"),a("OutboundLink")],1),e._v(" is supported by providing auto-configuration of the "),a("code",[e._v("spring-kafka")]),e._v(" project.")]),e._v(""),a("p",[e._v("Kafka configuration is controlled by external configuration properties in "),a("code",[e._v("spring.kafka.*")]),e._v(".\nFor example, you might declare the following section in "),a("code",[e._v("application.properties")]),e._v(":")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.kafka.bootstrap-servers=localhost:9092\nspring.kafka.consumer.group-id=myGroup\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n kafka:\n bootstrap-servers: "localhost:9092"\n consumer:\n group-id: "myGroup"\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("To create a topic on startup, add a bean of type "),a("code",[e._v("NewTopic")]),e._v("."),a("br"),e._v("If the topic already exists, the bean is ignored.")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("See "),a("a",{attrs:{href:"https://github.com/spring-projects/spring-boot/tree/v2.6.4/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java",target:"_blank",rel:"noopener noreferrer"}},[a("code",[e._v("KafkaProperties")]),a("OutboundLink")],1),e._v(" for more supported options.")]),e._v(""),a("h3",{attrs:{id:"_3-1-sending-a-message"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_3-1-sending-a-message"}},[e._v("#")]),e._v(" 3.1. Sending a Message")]),e._v(""),a("p",[e._v("Spring’s "),a("code",[e._v("KafkaTemplate")]),e._v(" is auto-configured, and you can autowire it directly in your own beans, as shown in the following example:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.springframework.kafka.core.KafkaTemplate;\nimport org.springframework.stereotype.Component;\n\n@Component\npublic class MyBean {\n\n private final KafkaTemplate<String, String> kafkaTemplate;\n\n public MyBean(KafkaTemplate<String, String> kafkaTemplate) {\n this.kafkaTemplate = kafkaTemplate;\n }\n\n // ...\n\n public void someMethod() {\n this.kafkaTemplate.send("someTopic", "Hello");\n }\n\n}\n\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("If the property "),a("code",[e._v("spring.kafka.producer.transaction-id-prefix")]),e._v(" is defined, a "),a("code",[e._v("KafkaTransactionManager")]),e._v(" is automatically configured."),a("br"),e._v("Also, if a "),a("code",[e._v("RecordMessageConverter")]),e._v(" bean is defined, it is automatically associated to the auto-configured "),a("code",[e._v("KafkaTemplate")]),e._v(".")])])]),e._v(""),a("tbody")]),e._v(""),a("h3",{attrs:{id:"_3-2-receiving-a-message"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_3-2-receiving-a-message"}},[e._v("#")]),e._v(" 3.2. Receiving a Message")]),e._v(""),a("p",[e._v("When the Apache Kafka infrastructure is present, any bean can be annotated with "),a("code",[e._v("@KafkaListener")]),e._v(" to create a listener endpoint.\nIf no "),a("code",[e._v("KafkaListenerContainerFactory")]),e._v(" has been defined, a default one is automatically configured with keys defined in "),a("code",[e._v("spring.kafka.listener.*")]),e._v(".")]),e._v(""),a("p",[e._v("The following component creates a listener endpoint on the "),a("code",[e._v("someTopic")]),e._v(" topic:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.springframework.kafka.annotation.KafkaListener;\nimport org.springframework.stereotype.Component;\n\n@Component\npublic class MyBean {\n\n @KafkaListener(topics = "someTopic")\n public void processMessage(String content) {\n // ...\n }\n\n}\n\n')])])]),a("p",[e._v("If a "),a("code",[e._v("KafkaTransactionManager")]),e._v(" bean is defined, it is automatically associated to the container factory.\nSimilarly, if a "),a("code",[e._v("RecordFilterStrategy")]),e._v(", "),a("code",[e._v("CommonErrorHandler")]),e._v(", "),a("code",[e._v("AfterRollbackProcessor")]),e._v(" or "),a("code",[e._v("ConsumerAwareRebalanceListener")]),e._v(" bean is defined, it is automatically associated to the default factory.")]),e._v(""),a("p",[e._v("Depending on the listener type, a "),a("code",[e._v("RecordMessageConverter")]),e._v(" or "),a("code",[e._v("BatchMessageConverter")]),e._v(" bean is associated to the default factory.\nIf only a "),a("code",[e._v("RecordMessageConverter")]),e._v(" bean is present for a batch listener, it is wrapped in a "),a("code",[e._v("BatchMessageConverter")]),e._v(".")]),e._v(""),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("A custom "),a("code",[e._v("ChainedKafkaTransactionManager")]),e._v(" must be marked "),a("code",[e._v("@Primary")]),e._v(" as it usually references the auto-configured "),a("code",[e._v("KafkaTransactionManager")]),e._v(" bean.")])])]),e._v(""),a("tbody")]),e._v(""),a("h3",{attrs:{id:"_3-3-kafka-streams"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_3-3-kafka-streams"}},[e._v("#")]),e._v(" 3.3. Kafka Streams")]),e._v(""),a("p",[e._v("Spring for Apache Kafka provides a factory bean to create a "),a("code",[e._v("StreamsBuilder")]),e._v(" object and manage the lifecycle of its streams.\nSpring Boot auto-configures the required "),a("code",[e._v("KafkaStreamsConfiguration")]),e._v(" bean as long as "),a("code",[e._v("kafka-streams")]),e._v(" is on the classpath and Kafka Streams is enabled by the "),a("code",[e._v("@EnableKafkaStreams")]),e._v(" annotation.")]),e._v(""),a("p",[e._v("Enabling Kafka Streams means that the application id and bootstrap servers must be set.\nThe former can be configured using "),a("code",[e._v("spring.kafka.streams.application-id")]),e._v(", defaulting to "),a("code",[e._v("spring.application.name")]),e._v(" if not set.\nThe latter can be set globally or specifically overridden only for streams.")]),e._v(""),a("p",[e._v("Several additional properties are available using dedicated properties; other arbitrary Kafka properties can be set using the "),a("code",[e._v("spring.kafka.streams.properties")]),e._v(" namespace.\nSee also "),a("RouterLink",{attrs:{to:"/en/spring-boot/features.html#messaging.kafka.additional-properties"}},[e._v("features.html")]),e._v(" for more information.")],1),e._v(""),a("p",[e._v("To use the factory bean, wire "),a("code",[e._v("StreamsBuilder")]),e._v(" into your "),a("code",[e._v("@Bean")]),e._v(" as shown in the following example:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.apache.kafka.common.serialization.Serdes;\nimport org.apache.kafka.streams.KeyValue;\nimport org.apache.kafka.streams.StreamsBuilder;\nimport org.apache.kafka.streams.kstream.KStream;\nimport org.apache.kafka.streams.kstream.Produced;\n\nimport org.springframework.context.annotation.Bean;\nimport org.springframework.context.annotation.Configuration;\nimport org.springframework.kafka.annotation.EnableKafkaStreams;\nimport org.springframework.kafka.support.serializer.JsonSerde;\n\n@Configuration(proxyBeanMethods = false)\n@EnableKafkaStreams\npublic class MyKafkaStreamsConfiguration {\n\n @Bean\n public KStream<Integer, String> kStream(StreamsBuilder streamsBuilder) {\n KStream<Integer, String> stream = streamsBuilder.stream("ks1In");\n stream.map(this::uppercaseValue).to("ks1Out", Produced.with(Serdes.Integer(), new JsonSerde<>()));\n return stream;\n }\n\n private KeyValue<Integer, String> uppercaseValue(Integer key, String value) {\n return new KeyValue<>(key, value.toUpperCase());\n }\n\n}\n\n')])])]),a("p",[e._v("By default, the streams managed by the "),a("code",[e._v("StreamBuilder")]),e._v(" object it creates are started automatically.\nYou can customize this behavior using the "),a("code",[e._v("spring.kafka.streams.auto-startup")]),e._v(" property.")]),e._v(""),a("h3",{attrs:{id:"_3-4-additional-kafka-properties"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_3-4-additional-kafka-properties"}},[e._v("#")]),e._v(" 3.4. Additional Kafka Properties")]),e._v(""),a("p",[e._v("The properties supported by auto configuration are shown in the "),a("RouterLink",{attrs:{to:"/en/spring-boot/application-properties.html#appendix.application-properties.integration"}},[e._v("“Integration Properties”")]),e._v(" section of the Appendix.\nNote that, for the most part, these properties (hyphenated or camelCase) map directly to the Apache Kafka dotted properties.\nSee the Apache Kafka documentation for details.")],1),e._v(""),a("p",[e._v("The first few of these properties apply to all components (producers, consumers, admins, and streams) but can be specified at the component level if you wish to use different values.\nApache Kafka designates properties with an importance of HIGH, MEDIUM, or LOW.\nSpring Boot auto-configuration supports all HIGH importance properties, some selected MEDIUM and LOW properties, and any properties that do not have a default value.")]),e._v(""),a("p",[e._v("Only a subset of the properties supported by Kafka are available directly through the "),a("code",[e._v("KafkaProperties")]),e._v(" class.\nIf you wish to configure the producer or consumer with additional properties that are not directly supported, use the following properties:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.kafka.properties[prop.one]=first\nspring.kafka.admin.properties[prop.two]=second\nspring.kafka.consumer.properties[prop.three]=third\nspring.kafka.producer.properties[prop.four]=fourth\nspring.kafka.streams.properties[prop.five]=fifth\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n kafka:\n properties:\n "[prop.one]": "first"\n admin:\n properties:\n "[prop.two]": "second"\n consumer:\n properties:\n "[prop.three]": "third"\n producer:\n properties:\n "[prop.four]": "fourth"\n streams:\n properties:\n "[prop.five]": "fifth"\n')])])]),a("p",[e._v("This sets the common "),a("code",[e._v("prop.one")]),e._v(" Kafka property to "),a("code",[e._v("first")]),e._v(" (applies to producers, consumers and admins), the "),a("code",[e._v("prop.two")]),e._v(" admin property to "),a("code",[e._v("second")]),e._v(", the "),a("code",[e._v("prop.three")]),e._v(" consumer property to "),a("code",[e._v("third")]),e._v(", the "),a("code",[e._v("prop.four")]),e._v(" producer property to "),a("code",[e._v("fourth")]),e._v(" and the "),a("code",[e._v("prop.five")]),e._v(" streams property to "),a("code",[e._v("fifth")]),e._v(".")]),e._v(""),a("p",[e._v("You can also configure the Spring Kafka "),a("code",[e._v("JsonDeserializer")]),e._v(" as follows:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer\nspring.kafka.consumer.properties[spring.json.value.default.type]=com.example.Invoice\nspring.kafka.consumer.properties[spring.json.trusted.packages]=com.example.main,com.example.another\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n kafka:\n consumer:\n value-deserializer: "org.springframework.kafka.support.serializer.JsonDeserializer"\n properties:\n "[spring.json.value.default.type]": "com.example.Invoice"\n "[spring.json.trusted.packages]": "com.example.main,com.example.another"\n')])])]),a("p",[e._v("Similarly, you can disable the "),a("code",[e._v("JsonSerializer")]),e._v(" default behavior of sending type information in headers:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer\nspring.kafka.producer.properties[spring.json.add.type.headers]=false\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n kafka:\n producer:\n value-serializer: "org.springframework.kafka.support.serializer.JsonSerializer"\n properties:\n "[spring.json.add.type.headers]": false\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("Properties set in this way override any configuration item that Spring Boot explicitly supports.")])])]),e._v(""),a("tbody")]),e._v(""),a("h3",{attrs:{id:"_3-5-testing-with-embedded-kafka"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_3-5-testing-with-embedded-kafka"}},[e._v("#")]),e._v(" 3.5. Testing with Embedded Kafka")]),e._v(""),a("p",[e._v("Spring for Apache Kafka provides a convenient way to test projects with an embedded Apache Kafka broker.\nTo use this feature, annotate a test class with "),a("code",[e._v("@EmbeddedKafka")]),e._v(" from the "),a("code",[e._v("spring-kafka-test")]),e._v(" module.\nFor more information, please see the Spring for Apache Kafka "),a("a",{attrs:{href:"https://docs.spring.io/spring-kafka/docs/2.8.3/reference/html/#embedded-kafka-annotation",target:"_blank",rel:"noopener noreferrer"}},[e._v("reference manual"),a("OutboundLink")],1),e._v(".")]),e._v(""),a("p",[e._v("To make Spring Boot auto-configuration work with the aforementioned embedded Apache Kafka broker, you need to remap a system property for embedded broker addresses (populated by the "),a("code",[e._v("EmbeddedKafkaBroker")]),e._v(") into the Spring Boot configuration property for Apache Kafka.\nThere are several ways to do that:")]),e._v(""),a("ul",[a("li",[e._v("Provide a system property to map embedded broker addresses into "),a("code",[e._v("spring.kafka.bootstrap-servers")]),e._v(" in the test class:")])]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('static {\n System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers");\n}\n\n')])])]),a("ul",[a("li",[e._v("Configure a property name on the "),a("code",[e._v("@EmbeddedKafka")]),e._v(" annotation:")])]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import org.springframework.boot.test.context.SpringBootTest;\nimport org.springframework.kafka.test.context.EmbeddedKafka;\n\n@SpringBootTest\n@EmbeddedKafka(topics = "someTopic", bootstrapServersProperty = "spring.kafka.bootstrap-servers")\nclass MyTest {\n\n // ...\n\n}\n\n')])])]),a("ul",[a("li",[e._v("Use a placeholder in configuration properties:")])]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n kafka:\n bootstrap-servers: "${spring.embedded.kafka.brokers}"\n')])])]),a("h2",{attrs:{id:"_4-rsocket"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-rsocket"}},[e._v("#")]),e._v(" 4. RSocket")]),e._v(""),a("p",[a("a",{attrs:{href:"https://rsocket.io",target:"_blank",rel:"noopener noreferrer"}},[e._v("RSocket"),a("OutboundLink")],1),e._v(" is a binary protocol for use on byte stream transports.\nIt enables symmetric interaction models through async message passing over a single connection.")]),e._v(""),a("p",[e._v("The "),a("code",[e._v("spring-messaging")]),e._v(" module of the Spring Framework provides support for RSocket requesters and responders, both on the client and on the server side.\nSee the "),a("a",{attrs:{href:"https://docs.spring.io/spring-framework/docs/5.3.16/reference/html/web-reactive.html#rsocket-spring",target:"_blank",rel:"noopener noreferrer"}},[e._v("RSocket section"),a("OutboundLink")],1),e._v(" of the Spring Framework reference for more details, including an overview of the RSocket protocol.")]),e._v(""),a("h3",{attrs:{id:"_4-1-rsocket-strategies-auto-configuration"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-rsocket-strategies-auto-configuration"}},[e._v("#")]),e._v(" 4.1. RSocket Strategies Auto-configuration")]),e._v(""),a("p",[e._v("Spring Boot auto-configures an "),a("code",[e._v("RSocketStrategies")]),e._v(" bean that provides all the required infrastructure for encoding and decoding RSocket payloads.\nBy default, the auto-configuration will try to configure the following (in order):")]),e._v(""),a("ol",[a("li",[a("p",[a("a",{attrs:{href:"https://cbor.io/",target:"_blank",rel:"noopener noreferrer"}},[e._v("CBOR"),a("OutboundLink")],1),e._v(" codecs with Jackson")])]),e._v(""),a("li",[a("p",[e._v("JSON codecs with Jackson")])])]),e._v(""),a("p",[e._v("The "),a("code",[e._v("spring-boot-starter-rsocket")]),e._v(" starter provides both dependencies.\nSee the "),a("RouterLink",{attrs:{to:"/en/spring-boot/features.html#features.json.jackson"}},[e._v("Jackson support section")]),e._v(" to know more about customization possibilities.")],1),e._v(""),a("p",[e._v("Developers can customize the "),a("code",[e._v("RSocketStrategies")]),e._v(" component by creating beans that implement the "),a("code",[e._v("RSocketStrategiesCustomizer")]),e._v(" interface.\nNote that their "),a("code",[e._v("@Order")]),e._v(" is important, as it determines the order of codecs.")]),e._v(""),a("h3",{attrs:{id:"_4-2-rsocket-server-auto-configuration"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-2-rsocket-server-auto-configuration"}},[e._v("#")]),e._v(" 4.2. RSocket server Auto-configuration")]),e._v(""),a("p",[e._v("Spring Boot provides RSocket server auto-configuration.\nThe required dependencies are provided by the "),a("code",[e._v("spring-boot-starter-rsocket")]),e._v(".")]),e._v(""),a("p",[e._v("Spring Boot allows exposing RSocket over WebSocket from a WebFlux server, or standing up an independent RSocket server.\nThis depends on the type of application and its configuration.")]),e._v(""),a("p",[e._v("For WebFlux application (that is of type "),a("code",[e._v("WebApplicationType.REACTIVE")]),e._v("), the RSocket server will be plugged into the Web Server only if the following properties match:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.rsocket.server.mapping-path=/rsocket\nspring.rsocket.server.transport=websocket\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n rsocket:\n server:\n mapping-path: "/rsocket"\n transport: "websocket"\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(""),a("th",[e._v("Plugging RSocket into a web server is only supported with Reactor Netty, as RSocket itself is built with that library.")])])]),e._v(""),a("tbody")]),e._v(""),a("p",[e._v("Alternatively, an RSocket TCP or websocket server is started as an independent, embedded server.\nBesides the dependency requirements, the only required configuration is to define a port for that server:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.rsocket.server.port=9898\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring:\n rsocket:\n server:\n port: 9898\n")])])]),a("h3",{attrs:{id:"_4-3-spring-messaging-rsocket-support"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-3-spring-messaging-rsocket-support"}},[e._v("#")]),e._v(" 4.3. Spring Messaging RSocket support")]),e._v(""),a("p",[e._v("Spring Boot will auto-configure the Spring Messaging infrastructure for RSocket.")]),e._v(""),a("p",[e._v("This means that Spring Boot will create a "),a("code",[e._v("RSocketMessageHandler")]),e._v(" bean that will handle RSocket requests to your application.")]),e._v(""),a("h3",{attrs:{id:"_4-4-calling-rsocket-services-with-rsocketrequester"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-4-calling-rsocket-services-with-rsocketrequester"}},[e._v("#")]),e._v(" 4.4. Calling RSocket Services with RSocketRequester")]),e._v(""),a("p",[e._v("Once the "),a("code",[e._v("RSocket")]),e._v(" channel is established between server and client, any party can send or receive requests to the other.")]),e._v(""),a("p",[e._v("As a server, you can get injected with an "),a("code",[e._v("RSocketRequester")]),e._v(" instance on any handler method of an RSocket "),a("code",[e._v("@Controller")]),e._v(".\nAs a client, you need to configure and establish an RSocket connection first.\nSpring Boot auto-configures an "),a("code",[e._v("RSocketRequester.Builder")]),e._v(" for such cases with the expected codecs and applies any "),a("code",[e._v("RSocketConnectorConfigurer")]),e._v(" bean.")]),e._v(""),a("p",[e._v("The "),a("code",[e._v("RSocketRequester.Builder")]),e._v(" instance is a prototype bean, meaning each injection point will provide you with a new instance .\nThis is done on purpose since this builder is stateful and you should not create requesters with different setups using the same instance.")]),e._v(""),a("p",[e._v("The following code shows a typical example:")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('import reactor.core.publisher.Mono;\n\nimport org.springframework.messaging.rsocket.RSocketRequester;\nimport org.springframework.stereotype.Service;\n\n@Service\npublic class MyService {\n\n private final RSocketRequester rsocketRequester;\n\n public MyService(RSocketRequester.Builder rsocketRequesterBuilder) {\n this.rsocketRequester = rsocketRequesterBuilder.tcp("example.org", 9898);\n }\n\n public Mono<User> someRSocketCall(String name) {\n return this.rsocketRequester.route("user").data(name).retrieveMono(User.class);\n }\n\n}\n\n')])])]),a("h2",{attrs:{id:"_5-spring-integration"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_5-spring-integration"}},[e._v("#")]),e._v(" 5. Spring Integration")]),e._v(""),a("p",[e._v("Spring Boot offers several conveniences for working with "),a("a",{attrs:{href:"https://spring.io/projects/spring-integration",target:"_blank",rel:"noopener noreferrer"}},[e._v("Spring Integration"),a("OutboundLink")],1),e._v(", including the "),a("code",[e._v("spring-boot-starter-integration")]),e._v(" “Starter”.\nSpring Integration provides abstractions over messaging and also other transports such as HTTP, TCP, and others.\nIf Spring Integration is available on your classpath, it is initialized through the "),a("code",[e._v("@EnableIntegration")]),e._v(" annotation.")]),e._v(""),a("p",[e._v("Spring Integration polling logic relies "),a("RouterLink",{attrs:{to:"/en/spring-boot/features.html#features.task-execution-and-scheduling"}},[e._v("on the auto-configured "),a("code",[e._v("TaskScheduler")])]),e._v(".\nThe default "),a("code",[e._v("PollerMetadata")]),e._v(" (poll unbounded number of messages every second) can be customized with "),a("code",[e._v("spring.integration.poller.*")]),e._v(" configuration properties.")],1),e._v(""),a("p",[e._v("Spring Boot also configures some features that are triggered by the presence of additional Spring Integration modules.\nIf "),a("code",[e._v("spring-integration-jmx")]),e._v(" is also on the classpath, message processing statistics are published over JMX.\nIf "),a("code",[e._v("spring-integration-jdbc")]),e._v(" is available, the default database schema can be created on startup, as shown in the following line:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("spring.integration.jdbc.initialize-schema=always\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('spring:\n integration:\n jdbc:\n initialize-schema: "always"\n')])])]),a("p",[e._v("If "),a("code",[e._v("spring-integration-rsocket")]),e._v(" is available, developers can configure an RSocket server using "),a("code",[e._v('"spring.rsocket.server.*"')]),e._v(" properties and let it use "),a("code",[e._v("IntegrationRSocketEndpoint")]),e._v(" or "),a("code",[e._v("RSocketOutboundGateway")]),e._v(" components to handle incoming RSocket messages.\nThis infrastructure can handle Spring Integration RSocket channel adapters and "),a("code",[e._v("@MessageMapping")]),e._v(" handlers (given "),a("code",[e._v('"spring.integration.rsocket.server.message-mapping-enabled"')]),e._v(" is configured).")]),e._v(""),a("p",[e._v("Spring Boot can also auto-configure an "),a("code",[e._v("ClientRSocketConnector")]),e._v(" using configuration properties:")]),e._v(""),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("# Connecting to a RSocket server over TCP\nspring.integration.rsocket.client.host=example.org\nspring.integration.rsocket.client.port=9898\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('# Connecting to a RSocket server over TCP\nspring:\n integration:\n rsocket:\n client:\n host: "example.org"\n port: 9898\n')])])]),a("p",[e._v("Properties")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("# Connecting to a RSocket Server over WebSocket\nspring.integration.rsocket.client.uri=ws://example.org\n")])])]),a("p",[e._v("Yaml")]),e._v(""),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('# Connecting to a RSocket Server over WebSocket\nspring:\n integration:\n rsocket:\n client:\n uri: "ws://example.org"\n')])])]),a("p",[e._v("See the "),a("a",{attrs:{href:"https://github.com/spring-projects/spring-boot/tree/v2.6.4/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/integration/IntegrationAutoConfiguration.java",target:"_blank",rel:"noopener noreferrer"}},[a("code",[e._v("IntegrationAutoConfiguration")]),a("OutboundLink")],1),e._v(" and "),a("a",{attrs:{href:"https://github.com/spring-projects/spring-boot/tree/v2.6.4/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/integration/IntegrationProperties.java",target:"_blank",rel:"noopener noreferrer"}},[a("code",[e._v("IntegrationProperties")]),a("OutboundLink")],1),e._v(" classes for more details.")]),e._v(""),a("h2",{attrs:{id:"_6-what-to-read-next"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_6-what-to-read-next"}},[e._v("#")]),e._v(" 6. What to Read Next")]),e._v(""),a("p",[e._v("The next section describes how to enable "),a("RouterLink",{attrs:{to:"/en/spring-boot/io.html#io"}},[e._v("IO capabilities")]),e._v(" in your application.\nYou can read about "),a("RouterLink",{attrs:{to:"/en/spring-boot/io.html#io.caching"}},[e._v("caching")]),e._v(", "),a("RouterLink",{attrs:{to:"/en/spring-boot/io.html#io.email"}},[e._v("mail")]),e._v(", "),a("RouterLink",{attrs:{to:"/en/spring-boot/io.html#io.validation"}},[e._v("validation")]),e._v(", "),a("RouterLink",{attrs:{to:"/en/spring-boot/io.html#io.rest-client"}},[e._v("rest clients")]),e._v(" and more in this section.")],1)])}),[],!1,null,null,null);t.default=n.exports}}]);
