1、概念：SpringBoot 整合消息服務2、具體內容對于異步消息組件在實際的應用之中會有兩類：· JMS：代表作就是 ActiveMQ，但是其性能不高，因為其是用 java 程序實現的；· AMQP：直接利用協議實現的消息組件，其大眾代表作：RabbitMQ，高性能代表作：Kafka。2.1、SpringBoot 整合 ActiveMQ1、 如果要想在項目之中去使用 ActiveMQ 組件，則應該為項目添加依賴支持庫，修改 pom.xml 配置文件：<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-activemq</artifactId> </dependency>2、 修改 application.properties 配置文件進行 activemq 的配置；spring.jms.pub-sub-domain=false spring.activemq.user=admin spring.activemq.password=1234 spring.activemq.broker-url=tcp://59.115.158.145:616163、 隨后定義一個消息的消費者，消費者主要是進行一個監聽控制，在 SpringBoot 里面可以直接利用 注解@JmsListener進行監聽：package com.microboot.consumer;import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Service;@Service public class MessageConsumerService {@JmsListener(destination="study.msg.queue")public void receiveMessage(String text) { // 進行消息接收處理System.err.println("【*** 接收消息 ***】" + text);} } 4、 隨后建立消息的發送者服務，一般而言如果進行消息的發送往往會準備出一個業務接口來：package com.microboot.producer;public interface IMessageProducerService {public void sendMessage(String msg) ; } 5、 隨后建立一個配置程序類，定義 ActiveMQ 的消息發送模版處理類：package com.microboot.config;import javax.jms.Queue;import org.apache.activemq.command.ActiveMQQueue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jms.annotation.EnableJms;@Configuration @EnableJms public class ActiveMQConfig {@Beanpublic Queue queue() {return new ActiveMQQueue("study.msg.queue");} } 6、 創建消息發送的子類實現消息發送處理：package com.microboot.producer;import javax.annotation.Resource; import javax.jms.Queue;import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.stereotype.Service;@Service public class MessageProducerServiceImpl implements IMessageProducerService {@Resourceprivate JmsMessagingTemplate jmsMessagingTemplate;@Resourceprivate Queue queue;@Overridepublic void sendMessage(String msg) {this.jmsMessagingTemplate.convertAndSend(this.queue, msg);}} 7、 編寫測試類來觀察消息的處理：package com.microboot.test;import javax.annotation.Resource;import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.web.WebAppConfiguration;import com.microboot.StartSpringBootMain; import com.microboot.producer.IMessageProducerService;@SpringBootTest(classes = StartSpringBootMain.class) @RunWith(SpringJUnit4ClassRunner.class) @WebAppConfiguration public class TestActiveMQ {@Resourceprivate IMessageProducerService messageProducer;@Testpublic void testSend() throws Exception {for (int x = 0; x < 10; x++) {this.messageProducer.sendMessage("study - " + x);}} }基于 SpringBoot 配置的 JMS 的組件訪問整體的處理十分簡單 2.2、SpringBoot 整合 RabbitMQ如果要進行 RabbitMQ 整合的時候一定要注意以下幾個概念：交換空間、虛擬主機、隊列信息。本次為了方便起見將項目分為 兩個：RabbitMQ-Consumer、RabbitMQ-Producer。1、 【兩個項目】將 rabbitmq 的依賴支持包拷貝到項目之中；<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId> </dependency>2、 【microboot-rabbitmq-producer、microboot-rabbitmq-consumer】修改 application.properties配置文件，追加 rabbitmq 的相關配置項：spring.rabbitmq.addresses=127.0.0.1 spring.rabbitmq.username=admin spring.rabbitmq.password=admin 3、 【microboot-rabbitmq-producer】建立一個消息的發送接口：package cn.study.microboot.producer;public interface IMessageProducerService {public void sendMessage(String msg) ; } 4、 【microboot-rabbitmq-producer】為了可以正常使用 RabbitMQ 進行消息處理，你還需要做一個消息生產配置類；package cn.study.microboot.config;import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;@Configuration public class ProducerConfig {public static final String EXCHANGE = "study.microboot.exchange"; // 交換空間名稱public static final String ROUTINGKEY = "study.microboot.routingkey"; // 設置路由keypublic static final String QUEUE_NAME = "study.microboot.queue"; // 隊列名稱@Beanpublic Binding bindingExchangeQueue(DirectExchange exchange,Queue queue) {return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY) ;}@Beanpublic DirectExchange getDirectExchange() { // 使用直連的模式return new DirectExchange(EXCHANGE, true, true);}@Beanpublic Queue queue() { // 要創建的隊列信息return new Queue(QUEUE_NAME);} } 5、 【microboot-rabbitmq-producer】創建消息服務的實現子類：package cn.study.microboot.producer.impl;import javax.annotation.Resource;import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.stereotype.Service;import cn.study.microboot.config.ProducerConfig; import cn.study.microboot.producer.IMessageProducerService;@Service public class MessageProducerServiceImpl implements IMessageProducerService {@Resourceprivate RabbitTemplate rabbitTemplate;@Overridepublic void sendMessage(String msg) {this.rabbitTemplate.convertAndSend(ProducerConfig.EXCHANGE,ProducerConfig.ROUTINGKEY, msg);}} 6、 【microboot-rabbitmq-consumer】依然需要做一個消費者的配置程序類，而這個程序類里面主要的目的依然是設置交換空間、 路由 KEY 等信息。package cn.study.microboot.config;import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;@Configuration public class ConsumerConfig {public static final String EXCHANGE = "study.microboot.exchange"; // 交換空間名稱public static final String ROUTINGKEY = "study.microboot.routingkey"; // 設置路由keypublic static final String QUEUE_NAME = "study.microboot.queue"; // 隊列名稱@Beanpublic Queue queue() { // 要創建的隊列信息return new Queue(QUEUE_NAME);}@Beanpublic DirectExchange getDirectExchange() { // 使用直連的模式return new DirectExchange(EXCHANGE, true, true);}@Beanpublic Binding bindingExchangeQueue(DirectExchange exchange,Queue queue) {return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY) ;} } 7、 【microboot-rabbitmq-consumer】實現監聽處理類：package cn.study.microboot.consumer;import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service;@Service public class MessageConsumerService {@RabbitListener(queues="study.microboot.queue")public void receiveMessage(String text) { // 進行消息接收處理System.err.println("【*** 接收消息 ***】" + text);} } 9、 【microboot-rabbitmq-consumer】編寫消息接收測試類，這里面不需要編寫代碼，只需要做一個休眠即可：package cn.study.microboot.test;import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.web.WebAppConfiguration;import cn.study.microboot.StartSpringBootMain;@SpringBootTest(classes = StartSpringBootMain.class) @RunWith(SpringJUnit4ClassRunner.class) @WebAppConfiguration public class AppTest {@Testpublic void testStart() throws Exception {Thread.sleep(Long.MAX_VALUE);} }整體進行項目開發之中整合的處理步驟還是簡單，但是千萬要注意，由于是第一次整合處理，所以將生產者與消費者的配置 類分開了，實際上這兩個類的作用是完全一樣的。 2.3、SpringBoot 整合 KafkaKafka 是現在最好的開源消息組件，其仿照 AMQP 協議操作，而且處理的性能也是最高的。本次使用已經配置好的 Kafka 服 務器，而且這臺服務器上使用了 kerberos 認證，所以應該首先準備好一個 jass 配置文件：1、 定義“kafka_client_jaas.conf”配置文件：KafkaClient {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="bob"password="bob-pwd"; }; 2、 為了方便進行項目的觀察， 本次依然準備出了兩個項目：生產者（ microboot-kafka-producer ）、 消 費 者 （microboot-kafka-consumer），隨后為這兩個項目添加 kafka 配置支持：<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency> 3、 【micorboot-kafka-consumer】修改 application.yml 配置文件，進行 kafka 配置項編寫：server:port: 80 spring:messages:basename: i18n/Messages,i18n/Pageskafka:bootstrap-servers:- kafka-single:9095template:default-topic: mldn-microbootconsumer:key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializergroup-id: group-1properties:sasl.mechanism: PLAINsecurity.protocol: SASL_PLAINTEXT 4、 【micorboot-kafka-consumer】建立一個 Kafka 的消息的消費程序類：package cn.study.microboot.consumer;import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Service;@Service public class MessageConsumerService {@KafkaListener(topics = {"study-microboot"})public void receiveMessage(ConsumerRecord<String, String> record) { // 進行消息接收處理System.err.println("【*** 接收消息 ***】key = " + record.key() + "、value = "+ record.value());} } 5、 【micorboot-kafka-consumer】隨后還需要修改 SpringBoot 的啟動程序類，追加 kerberos 配置：package cn.study.microboot;import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication // 啟動SpringBoot程序，而后自帶子包掃描 public class StartSpringBootMain {static {System.setProperty("java.security.auth.login.config","d:/kafka_client_jaas.conf"); // 表示系統環境屬性}public static void main(String[] args) throws Exception {SpringApplication.run(StartSpringBootMain.class, args);} } 6、 【microboot-kafka-producer】修改 application.yml 配置文件：server:port: 80 spring:messages:basename: i18n/Messages,i18n/Pageskafka:bootstrap-servers:- kafka-single:9095template:default-topic: mldn-microbootproducer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerproperties:sasl.mechanism: PLAINsecurity.protocol: SASL_PLAINTEXT 7、 【microboot-kafka-producer】定義消息發送的服務接口：package cn.study.microboot.producer;public interface IMessageProducerService {public void sendMessage(String msg) ; }package cn.study.microboot.service.impl;import javax.annotation.Resource;import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Service;import cn.study.microboot.service.IMessageProducerService;@Service public class MessageProducerServiceImpl implements IMessageProducerService {@Resourceprivate KafkaTemplate<String, String> kafkaTemplate;@Overridepublic void send(String msg) {this.kafkaTemplate.sendDefault("study-key", msg);}} 8、 【microboot-kafka-producer】修改程序啟動類：package cn.mldn.microboot;import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication // 啟動SpringBoot程序，而后自帶子包掃描 public class StartSpringBootMain {static {System.setProperty("java.security.auth.login.config","d:/kafka_client_jaas.conf"); // 表示系統環境屬性}public static void main(String[] args) throws Exception {SpringApplication.run(StartSpringBootMain.class, args);} } 9、 【microboot-kafka-producer】編寫消息發送的程序類：package cn.study.microboot;import javax.annotation.Resource;import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.web.WebAppConfiguration;import cn.study.microboot.service.IMessageProducerService; @SpringBootTest(classes = StartSpringBootMain.class) @RunWith(SpringJUnit4ClassRunner.class) @WebAppConfiguration public class TestMessageService {@Resourceprivate IMessageProducerService messageService;@Testpublic void testStart() throws Exception {for (int x = 0; x < 100; x++) {this.messageService.send("study - " + x);}} }

?

總結

以上是生活随笔為你收集整理的SpringBoot整合 ActiveMQ、SpringBoot整合RabbitMQ、SpringBoot整合Kafka的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。