SpringBoot3.4.0整合消息中间件Kafka和RabbitMQ

itbeien2025年3月3日大约 6 分钟

SpringBoot3.4.0整合消息中间件Kafka和RabbitMQ

上一篇文章基于Docker完成了SpringBoot3.4.0和Apache Pulsar4.0.1整合并实现了消息的发送和订阅，今天基于SpringBoot实现整合Kafka和RabbitMQ并实现发送订阅消息功能。Docker Desktop 4.36.0的安装流程见基于Docker-SpringBoot3.4.0集成Apache Pulsar4.0.1实现消息发布和订阅

完整代码在文章最后，如果觉得本篇文章对你有用，记得点赞、关注、收藏哦。你的支持是我持续更新的动力!

SpringBoot3专栏软件环境

JDK17.0.12
SpringBoot3.4.0
Kafka3.9.0
RabbitMQ4.0
Docker Desktop4.36.0 for Windows
IDEA2024.2.0.2
Win10专业版

我们先看本篇文章对应的项目结构，请看下图

1 基于Docker安装RabbitMQ

1.1 安装和启动RabbitMQ

# latest RabbitMQ 4.0.x
docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4.0-management

2 RabbitMQ整合Springboot3.4.0

2.1 pom.xml依赖配置

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>cn.itbeien</groupId>
        <artifactId>springboot3-labs-master</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <artifactId>springboot-rabbitmq</artifactId>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>
</project>

2.2 代码实现

2.2.1 配置信息

# application.properties
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

2.2.2 生产者代码

package cn.itbeien.rabbitmq.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;


/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Copyright© 2024 itbeien
 */
@Service
@Slf4j
public class ProducerService {

    private final AmqpAdmin amqpAdmin;

    private final AmqpTemplate amqpTemplate;

    @Bean
    public Queue itbeienQueue() {
        return new Queue("itbeienQueue");
    }

    public ProducerService(AmqpAdmin amqpAdmin ,AmqpTemplate amqpTemplate) {
        this.amqpAdmin = amqpAdmin;
        this.amqpTemplate = amqpTemplate;
    }

    public void sendMsg(){
        Message message = new Message("hello itbeien".getBytes());
        this.amqpTemplate.send("itbeienQueue",message);
        log.info("生产者发布消息...");
    }

}

2.2.3 消费者代码

package cn.itbeien.rabbitmq.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Copyright© 2024 itbeien
 */
@Service
@Slf4j
public class ConsumerService {

    @RabbitHandler
    public void process(@Payload String foo) {
        log.info(foo);
    }


    @RabbitListener(queues = "itbeienQueue")
    public void processMessage(String content) {
        log.info("消费者消费消息:{}",content);
    }
}

2.3 单元测试

2.3.1 单元测试代码

package cn.itbeien.rabbitmq.test;

import cn.itbeien.rabbitmq.service.ProducerService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;

/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Copyright© 2024 itbeien
 */
@SpringBootTest
public class RabbitMQApp {
    @Autowired
    private ProducerService producerService;

    @Test
   public void test() throws Exception{
        producerService.sendMsg();
        Thread.sleep(10000);
   }
}

2.3.2 单元测试结果

结合RabbitMQ完成消息的生产和消费，请看下图

3 基于Docker安装 Kafka

3.1 安装和启动Kafka

# 获取docker镜像
docker pull apache/kafka:3.9.0
# 启动运行kafka
docker run -p 9092:9092 apache/kafka:3.9.0

4 Kafka整合SpringBoot3.4.0

4.1 pom.xml依赖配置

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>cn.itbeien</groupId>
        <artifactId>springboot3-labs-master</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <artifactId>springboot-kafka</artifactId>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
    </dependencies>
</project>

4.2 代码实现

4.2.1 配置信息

#application.properties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=itbeienGroup
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=cn.itbeien.kafka.vo
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

4.2.2 生产者代码

package cn.itbeien.kafka.service;

import cn.itbeien.kafka.vo.SampleMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Copyright© 2024 itbeien
 */
@Service
@Slf4j
public class ProducerService {

    private final KafkaTemplate<Object, SampleMessage> kafkaTemplate;

    ProducerService(KafkaTemplate<Object, SampleMessage> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void send(SampleMessage message) {
        this.kafkaTemplate.send("itbeienTopic", message);
        log.info("Sent sample message [" + message + "]");
    }

}

4.2.3 消费者代码

package cn.itbeien.kafka.service;

import cn.itbeien.kafka.vo.SampleMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Copyright© 2024 itbeien
 */
@Service
@Slf4j
public class ConsumerService {
    private final List<SampleMessage> messages = new CopyOnWriteArrayList<>();

    @KafkaListener(topics = "itbeienTopic")
    void processMessage(SampleMessage message) {
        this.messages.add(message);
        log.info("Received sample message [" + message + "]");
    }

    public List<SampleMessage> getMessages() {
        return this.messages;
    }
}

4.3 单元测试

4.3.1 单元测试代码

package cn.itbeien.kafka.test;

import cn.itbeien.kafka.service.ProducerService;
import cn.itbeien.kafka.vo.SampleMessage;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Copyright© 2024 itbeien
 */
@SpringBootTest
public class KafkaApp {
    @Autowired
    private ProducerService producerService;

    @Test
    public void test() throws Exception{
        SampleMessage sampleMessage = new SampleMessage(1,"hello itbeien");
        producerService.send(sampleMessage);
        Thread.sleep(5000);
    }
}