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 里面支持有数据的副本操做，能够保证数据更加安全。