zoukankan      html  css  js  c++  java
  • SpringBoot系列八:SpringBoot整合消息服务(SpringBoot 整合 ActiveMQ、SpringBoot 整合 RabbitMQ、SpringBoot 整合 Kafka)

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

  • 相关阅读:
    程序从命令行接收多个数字,求和之后输出结果。
    动态规划(1)
    软件工程个人作业(4)
    冲刺2 01
    构建之法
    水王
    大道至简第七章读后感
    构建之法03
    团队冲刺第四天
    团队冲刺第三天
  • 原文地址:https://www.cnblogs.com/leeSmall/p/8721556.html
Copyright © 2011-2022 走看看