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

本文轉載自查看原文 2018-04-05 11:19 12318 SpringBoot

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

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

2、具體內容

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

· JMS：代表作就是 ActiveMQ，但是其性能不高，因為其是用 java 程序實現的；

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

2.1、SpringBoot 整合 ActiveMQ

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

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

2、修改 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 # 消息組件的連接主機信息

3、隨后定義一個消息的消費者，消費者主要是進行一個監聽控制，在 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);
    }
}

4、隨后建立消息的發送者服務，一般而言如果進行消息的發送往往會准備出一個業務接口來：

package cn.study.microboot.producer;

public interface IMessageProducerService {
    public void sendMessage(String msg) ;
}

5、隨后建立一個配置程序類，定義 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") ;
    }
}

6、創建消息發送的子類實現消息發送處理：

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);
    }

}

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

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.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.yml 配置文件，追加 rabbitmq 的相關配置項：

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

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"; // 設置路由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);
    }
}

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 {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Override
    public 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"; // 設置路由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) ;
    }
}

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);
    }
}

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

9、【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.3、SpringBoot 整合 Kafka

Kafka 是現在最好的開源消息組件，其仿照 AMQP 協議操作，而且處理的性能也是最高的。本次使用已經配置好的 Kafka 服務器，而且這台服務器上使用了 kerberos 認證，所以應該首先准備好一個 jass 配置文件：

1、定義“kafka_client_jaas.conf”配置文件：

KafkaClient {
 org.apache.kafka.common.security.plain.PlainLoginModule required
 username="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/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

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/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

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

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

3、總結

實際開發之中 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 里面支持有數據的副本操作，可以保證數據更加安全。

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 【SpringBoot系列5】SpringBoot整合RabbitMQ springboot系列八、springboot整合kafka springboot系列八、springboot整合kafka Springboot 整合 Rabbitmq消息隊列 SpringBoot+ActiveMQ(整合) springboot整合activemq SpringBoot整合ActiveMQ springboot與ActiveMQ整合 SpringBoot 整合 ActiveMq springboot整合activeMQ