SpringBoot 笔记八：SpringBoot整合消息服务（SpringBoot 整合 ActiveMQ、SpringBoot 整合 RabbitMQ、SpringBoot 整合 Kafka）-搜云库技术团队

IDEA2023.1.3破解,IDEA破解,IDEA 2023.1破解,最新IDEA激活码

声明：本文来源于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 里面支持有数据的副本操作，可以保证数据更加安全。

来源：https://www.cnblogs.com/leeSmall/category/1185488.html

文章永久链接：https://tech.souyunku.com/?p=13955

SpringBoot 笔记八：SpringBoot整合消息服务（SpringBoot 整合 ActiveMQ、SpringBoot 整合 RabbitMQ、SpringBoot 整合 Kafka）

相关推荐

评论抢沙发

学习必备：精选优质专栏

大前端WP主题更专业更方便

觉得文章有用就打赏一下文章作者

微信扫一扫打赏

微信扫一扫打赏

QQ咨询

回顶部

相关推荐

评论 抢沙发

学习必备：精选优质专栏

大前端WP主题 更专业 更方便

觉得文章有用就打赏一下文章作者

微信扫一扫打赏

微信扫一扫打赏

QQ咨询

回顶部

评论抢沙发

大前端WP主题更专业更方便