SpringBoot系列八：SpringBoot整合消息服務（SpringBoot 整合 ActiveMQ、SpringBoot 整合 RabbitMQ、SpringBoot 整合 Kafka）

聲明：本文來源於MLDN培訓視頻的課堂筆記，寫在這裏只是爲了方便查閱。

一、概念：SpringBoot 整合消息服務

二、具體內容

對於異步消息組件在實際的應用之中會有兩類：

· JMS：表明做就是 ActiveMQ，可是其性能不高，由於其是用 java 程序實現的；

· AMQP：直接利用協議實現的消息組件，其大衆表明做：RabbitMQ，高性能表明做：Kafka。

2.一、SpringBoot 整合 ActiveMQ

一、 若是要想在項目之中去使用 ActiveMQ 組件，則應該爲項目添加依賴支持庫，修改 pom.xml 配置文件：

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>

二、 修改 application.yml 配置文件進行 activemq 的配置；

server: port: 80 spring: messages: basename: i18n/Messages,i18n/Pages jms: pub-sub-domain: false # 配置消息的類型，若是是true則表示爲topic消息，若是爲false表示Queue消息 activemq: user: studyjava # 鏈接用戶名 password: hello # 鏈接密碼 broker-url: tcp://activemq-server:61616 # 消息組件的鏈接主機信息

三、 隨後定義一個消息的消費者，消費者主要是進行一個監聽控制，在 SpringBoot 裏面能夠直接利用註解@JmsListener進行監聽：

package cn.study.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); } }

 

四、 隨後創建消息的發送者服務，通常而言若是進行消息的發送每每會準備出一個業務接口來：

package cn.study.microboot.producer; public interface IMessageProducerService { public void sendMessage(String msg) ; }

五、 隨後創建一個配置程序類，定義 ActiveMQ 的消息發送模版處理類：

package cn.study.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 { @Bean public Queue queue() { return new ActiveMQQueue("study.msg.queue") ; } }

六、 建立消息發送的子類實現消息發送處理：

package cn.study.microboot.producer.impl; import javax.annotation.Resource; import javax.jms.Queue; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.stereotype.Service; import cn.study.microboot.producer.IMessageProducerService; @Service public class MessageProducerServiceImpl implements IMessageProducerService { @Resource private JmsMessagingTemplate jmsMessagingTemplate; @Resource private Queue queue; @Override public void sendMessage(String msg) { this.jmsMessagingTemplate.convertAndSend(this.queue, msg); } }

 

七、 編寫測試類來觀察消息的處理：

package cn.study.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 cn.study.microboot.StartSpringBootMain; import cn.study.microboot.producer.IMessageProducerService; @SpringBootTest(classes = StartSpringBootMain.class) @RunWith(SpringJUnit4ClassRunner.class) @WebAppConfiguration public class TestActiveMQ { @Resource private IMessageProducerService messageProducer; @Test public void testSend() throws Exception { for (int x = 0; x < 10; x++) { this.messageProducer.sendMessage("study - " + x); } } }

 

基於 SpringBoot 配置的 JMS 的組件訪問總體的處理十分簡單

2.二、SpringBoot 整合 RabbitMQ

若是要進行 RabbitMQ 整合的時候必定要注意如下幾個概念：交換空間、虛擬主機、隊列信息。本次爲了方便起見將項目分爲 兩個：RabbitMQ-Consumer、RabbitMQ-Producer。

一、 【兩個項目】將 rabbitmq 的依賴支持包拷貝到項目之中；

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

二、 【microboot-rabbitmq-producer、microboot-rabbitmq-consumer】修改 application.yml 配置文件，追加 rabbitmq 的相關配置項：

server: port: 80 spring: messages: basename: i18n/Messages,i18n/Pages rabbitmq: addresses: rabbitmq-server username: studyjava password: hello virtual-host: /

三、 【microboot-rabbitmq-producer】創建一個消息的發送接口：

package cn.study.microboot.producer; public interface IMessageProducerService { public void sendMessage(String msg) ; }

 

四、 【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"; // 設置路由key
    public static final String QUEUE_NAME = "study.microboot.queue"; // 隊列名稱
 @Bean public Binding bindingExchangeQueue(DirectExchange exchange,Queue queue) { return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY) ; } @Bean public DirectExchange getDirectExchange() { // 使用直連的模式
        return new DirectExchange(EXCHANGE, true, true); } @Bean public Queue queue() { // 要建立的隊列信息
        return new Queue(QUEUE_NAME); } }

 

五、 【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 { @Resource private RabbitTemplate rabbitTemplate; @Override public void sendMessage(String msg) { this.rabbitTemplate.convertAndSend(ProducerConfig.EXCHANGE, ProducerConfig.ROUTINGKEY, msg); } }

六、 【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"; // 設置路由key
    public static final String QUEUE_NAME = "study.microboot.queue"; // 隊列名稱
 @Bean public Queue queue() { // 要建立的隊列信息
        return new Queue(QUEUE_NAME); } @Bean public DirectExchange getDirectExchange() { // 使用直連的模式
        return new DirectExchange(EXCHANGE, true, true); } @Bean public Binding bindingExchangeQueue(DirectExchange exchange,Queue queue) { return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY) ; } }

七、 【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); } }

 

八、 【microboot-rabbitmq-producer】建立一個測試類實現消息的發送處理。

package cn.study.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 cn.study.microboot.StartSpringBootMain; import cn.study.microboot.producer.IMessageProducerService; @SpringBootTest(classes = StartSpringBootMain.class) @RunWith(SpringJUnit4ClassRunner.class) @WebAppConfiguration public class TestActiveMQ { @Resource private IMessageProducerService messageProducer; @Test public void testSend() throws Exception { for (int x = 0; x < 100; x++) { this.messageProducer.sendMessage("study - " + x); } } }

九、 【microboot-rabbitmq-consumer】編寫消息接收測試類，這裏面不須要編寫代碼，只須要作一個休眠便可：

package cn.study.microboot; 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; @SpringBootTest(classes = StartSpringBootMain.class) @RunWith(SpringJUnit4ClassRunner.class) @WebAppConfiguration public class AppTest { @Test public void testStart() throws Exception { Thread.sleep(Long.MAX_VALUE); } }

總體進行項目開發之中整合的處理步驟仍是簡單，可是千萬要注意，因爲是第一次整合處理，因此將生產者與消費者的配置 類分開了，實際上這兩個類的做用是徹底同樣的。

2.三、SpringBoot 整合 Kafka

Kafka 是如今最好的開源消息組件，其仿照 AMQP 協議操做，並且處理的性能也是最高的。本次使用已經配置好的 Kafka 服 務器，並且這臺服務器上使用了 kerberos 認證，因此應該首先準備好一個 jass 配置文件：

一、 定義「kafka_client_jaas.conf」配置文件：

KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="bob" password="bob-pwd"; }; 

二、 爲了方便進行項目的觀察， 本次依然準備出了兩個項目：生產者（ microboot-kafka-producer ）、 消 費 者 （microboot-kafka-consumer），隨後爲這兩個項目添加 kafka 配置支持：

<dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

三、 【micorboot-kafka-consumer】修改 application.yml 配置文件，進行 kafka 配置項編寫：

server: port: 80 spring: messages: basename: i18n/Messages,i18n/Pages kafka: bootstrap-servers: - kafka-single:9095 template: default-topic: mldn-microboot consumer: key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer group-id: group-1 properties: sasl.mechanism: PLAIN security.protocol: SASL_PLAINTEXT

四、 【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()); } }

五、 【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); } }

六、 【microboot-kafka-producer】修改 application.yml 配置文件：

server: port: 80 spring: messages: basename: i18n/Messages,i18n/Pages kafka: bootstrap-servers: - kafka-single:9095 template: default-topic: mldn-microboot producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer properties: sasl.mechanism: PLAIN security.protocol: SASL_PLAINTEXT

七、 【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 { @Resource private KafkaTemplate<String, String> kafkaTemplate; @Override public void send(String msg) { this.kafkaTemplate.sendDefault("study-key", msg); } }

八、 【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); } }

九、 【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 { @Resource private IMessageProducerService messageService; @Test public void testStart() throws Exception { for (int x = 0; x < 100; x++) { this.messageService.send("study - " + x); } } }

在使用 Kafka 進行數據處理的時候必定要記住，它速度快的主要緣由是採用的協議、處理的模式、零拷貝。

三、總結

實際開發之中 90%環境下經常使用的三個消息組件：ActiveMQ、RabbitMQ、Kafka 的所有定義都在此處，之後大家所從事的開發 裏面必定會有消息組件的身影。消息組件帶來的最直觀好處：數據緩衝，能夠保證消息不丟失。

面試題：請解釋一下 ActiveMQ 與 RabbitMQ 區別？

· ActiveMQ 使用的是 JMS 協議處理，因此性能比較差，在 ActiveMQ 裏面其組成比較簡單就是進行主題或者是隊列消息的 處理；

·RabbitMQ 使用的是 AMQP 處理，該處理屬於一種協議處理，因此處理的性能會比較高，在 RabbitMQ 裏面提供有 exchange、 queue、bind 的概念，全部的用戶提交的消息發送給 exchange，然後由 bind 綁定 exchange 與 queue，最後根據 routingkey 進行消息 的發送處理，利用這一律念能夠實現 fanout（廣播）、topic（主題）、direct（直連）的操做處理。同時在 Rabbitmq 之中還經過有虛 擬主機的概念，也就是說不一樣的虛擬主機能夠有本身獨立的用戶管理、空間管理。

面試題：請解釋一下 RabbitMQ 與 Kafka 關係？

· 使用最爲普遍性能也比較好的就是 RabbitMQ 組件，Rabbitmq 中的消息消費完就刪除，RabbitMQ 自己支持的集羣功能有 限，必須結合 HAProxy、Keepalived 纔可以實現負載均衡與 HA 技術；

· Kafka 採用零拷貝、批量讀取技術能夠實現高效的消息交互，Kafka 中的消息會保存兩天，同時提供有一個 offset 能夠實現 歷史消息的讀取，Kafka 直接支持有 HA 與負載均衡的支持，在 Kafka 裏面支持有數據的副本操做，能夠保證數據更加安全。