java微服务中消息队列处理中间件基础语法学习，零基础学习

🕗 发布于 2025-01-21 06:14 java 中间件消息队列处理

在 Java 微服务中，消息队列处理中间件可以帮助实现服务之间的异步通信、解耦和负载均衡。常用的 Java 消息队列工具包括 RabbitMQ、Apache Kafka 和 ActiveMQ。下面我将详细介绍这些消息队列工具在 Java 中的基础语法和使用方法。

1. RabbitMQ

RabbitMQ 是一个广泛使用的开源消息代理软件，支持多种协议（AMQP、MQTT、STOMP 等）。我们可以使用 Spring AMQP 来简化 RabbitMQ 的集成。

1.1 安装 RabbitMQ

在 Ubuntu 上安装 RabbitMQ

bash

sudo apt-get update
sudo apt-get install rabbitmq-server

启动 RabbitMQ 服务

bash

sudo systemctl start rabbitmq-server
sudo systemctl enable rabbitmq-server

启用管理插件

bash

sudo rabbitmq-plugins enable rabbitmq_management

访问管理界面:打开浏览器，访问 http://<your_server_ip>:15672，默认用户名和密码为 guest。

1.2 基本概念

Exchange: 消息的路由中心。
Queue: 存储消息的地方。
Binding: 将 Exchange 和 Queue 关联起来。
Routing Key: 路由键，用于匹配 Binding。
Producer: 发送消息的应用程序。
Consumer: 接收消息的应用程序。

1.3 示例代码

1.3.1 创建 Spring Boot 项目

使用 Spring Initializr 创建一个新的 Spring Boot 项目，并添加以下依赖：

Spring Web
Spring AMQP
Lombok (可选)

1.3.2 配置 RabbitMQ

在 application.yml 中配置 RabbitMQ 连接信息：

yaml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

1.3.3 生产者

创建一个生产者类来发送消息：

RabbitMQSender.java

package com.example.demo.rabbitmq;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class RabbitMQSender {

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(String message) {
        amqpTemplate.convertAndSend("my_queue", message);
        System.out.println("Sent message = " + message);
    }
}

1.3.4 消费者

创建一个消费者类来接收消息：

RabbitMQReceiver.java

package com.example.demo.rabbitmq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class RabbitMQReceiver {

    @RabbitListener(queues = "my_queue")
    public void receiveMessage(String message) {
        log.info("Received message = {}", message);
    }
}

1.3.5 控制器

创建一个控制器来触发消息发送：

RabbitMQController.java

package com.example.demo.controller;

import com.example.demo.rabbitmq.RabbitMQSender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class RabbitMQController {

    @Autowired
    private RabbitMQSender rabbitMQSender;

    @GetMapping("/send-rabbitmq-message")
    public String sendMessage(@RequestParam String message) {
        rabbitMQSender.send(message);
        return "Message sent successfully";
    }
}

2. Apache Kafka

Apache Kafka 是一个分布式流处理平台，常用于实时数据管道和流应用程序。我们可以使用 Spring Kafka 来简化 Kafka 的集成。

2.1 安装 Kafka

下载并解压 Kafka

bash

wget https://downloads.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz
tar -xzf kafka_2.13-3.0.0.tgz
cd kafka_2.13-3.0.0

启动 Zookeeper:

bash

bin/zookeeper-server-start.sh config/zookeeper.properties

启动 Kafka 服务器

bash

bin/kafka-server-start.sh config/server.properties

2.2 基本概念

Broker: Kafka 集群中的节点。
Topic: 数据分类的主题。
Partition: Topic 的分区。
Producer: 发送消息的应用程序。
Consumer: 接收消息的应用程序。
Consumer Group: 消费者的组。

2.3 示例代码

2.3.1 创建 Spring Boot 项目

使用 Spring Initializr 创建一个新的 Spring Boot 项目，并添加以下依赖：

Spring Web
Spring for Apache Kafka
Lombok (可选)

2.3.2 配置 Kafka

在 application.yml 中配置 Kafka 连接信息：

yaml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

2.3.3 创建 Topic

在 Kafka 中创建一个主题 test-topic:

bash

bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1

2.3.4 生产者

创建一个生产者类来发送消息：

KafkaProducer.java

package com.example.demo.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String message) {
        kafkaTemplate.send("test-topic", message);
        System.out.println("Sent message = " + message);
    }
}

2.3.5 消费者

创建一个消费者类来接收消息：

KafkaConsumer.java

package com.example.demo.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class KafkaConsumer {

    @KafkaListener(topics = "test-topic", groupId = "my-group")
    public void consume(String message) {
        log.info("Consumed message = {}", message);
    }
}

2.3.6 控制器

创建一个控制器来触发消息发送：

KafkaController.java

package com.example.demo.controller;

import com.example.demo.kafka.KafkaProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class KafkaController {

    @Autowired
    private KafkaProducer kafkaProducer;

    @GetMapping("/send-kafka-message")
    public String sendMessage(@RequestParam String message) {
        kafkaProducer.sendMessage(message);
        return "Message sent successfully";
    }
}

3. ActiveMQ

ActiveMQ 是另一个流行的开源消息代理，支持 JMS 协议。我们可以使用 Spring JMS 来简化 ActiveMQ 的集成。

3.1 安装 ActiveMQ

下载并解压 ActiveMQ

bash

wget https://downloads.apache.org/activemq/5.16.5/apache-activemq-5.16.5-bin.tar.gz
tar -xzf apache-activemq-5.16.5-bin.tar.gz
cd apache-activemq-5.16.5

启动 ActiveMQ 服务:

bin/activemq start

访问管理界面:

打开浏览器，访问 http://<your_server_ip>:8161/admin，默认用户名和密码为 admin 和 admin。

3.2 基本概念

Broker: 消息代理。
Queue: 存储消息的地方。
Topic: 订阅模式的消息通道。
Producer: 发送消息的应用程序。
Consumer: 接收消息的应用程序。

3.3 示例代码

3.3.1 创建 Spring Boot 项目

使用 Spring Initializr 创建一个新的 Spring Boot 项目，并添加以下依赖：

Spring Web
Spring for Apache ActiveMQ
Lombok (可选)

3.3.2 配置 ActiveMQ

在 application.yml 中配置 ActiveMQ 连接信息：

yaml

spring:
  activemq:
    broker-url: tcp://localhost:61616
    user: admin
    password: admin

3.3.3 生产者

创建一个生产者类来发送消息：

ActiveMQProducer.java

package com.example.demo.activemq;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

@Service
public class ActiveMQProducer {

    @Autowired
    private JmsTemplate jmsTemplate;

    public void sendMessage(String message) {
        jmsTemplate.convertAndSend("my_queue", message);
        System.out.println("Sent message = " + message);
    }
}

3.3.4 消费者

创建一个消费者类来接收消息：

ActiveMQConsumer.java

package com.example.demo.activemq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class ActiveMQConsumer {

    @JmsListener(destination = "my_queue")
    public void receiveMessage(String message) {
        log.info("Received message = {}", message);
    }
}

3.3.5 控制器

创建一个控制器来触发消息发送：

ActiveMQController.java

package com.example.demo.controller;

import com.example.demo.activemq.ActiveMQProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class ActiveMQController {

    @Autowired
    private ActiveMQProducer activeMQProducer;

    @GetMapping("/send-activemq-message")
    public String sendMessage(@RequestParam String message) {
        activeMQProducer.sendMessage(message);
        return "Message sent successfully";
    }
}

4.RabbitMQ、Apache Kafka 和 ActiveMQ 在微服务架构中作为消息队列处理工具的优缺点对比。

1. RabbitMQ

优点

成熟稳定: 已经存在很长时间，社区支持强大。
丰富的协议支持: 支持多种协议（AMQP、MQTT、STOMP 等），灵活性高。
可靠的消息传递: 提供持久化和确认机制，确保消息不丢失。
易于管理: 提供强大的管理界面和监控工具。
广泛使用: 社区活跃，文档丰富。

缺点

性能限制: 相比 Kafka，在大规模数据处理方面可能稍逊一筹。
复杂性: 配置和管理相对复杂，尤其是在高级场景下。
许可证: 使用 Erlang VM，可能会增加一些学习成本。

2. Apache Kafka

优点

高性能: 设计用于高吞吐量和低延迟，适合大规模数据处理。
可扩展性强: 容易水平扩展，支持多副本和分区。
持久化能力强: 默认情况下所有消息都会被持久化到磁盘。
容错性好: 内置了复制机制，确保数据安全。
流处理能力: 内置流处理功能，支持实时数据处理。

缺点

配置复杂: 初次配置和调优较为复杂。
资源消耗大: 需要较多的内存和磁盘空间。
依赖 Zookeeper: 需要额外部署和维护 Zookeeper。
学习曲线陡峭: 对于新手来说，理解其内部工作原理需要一定时间。

3. ActiveMQ

优点

成熟的 JMS 实现: 符合 JMS 规范，适用于 Java 应用程序。
支持多种传输协议: 支持 TCP、UDP、HTTP 等多种传输协议。
广泛的集成: 与 Spring 等框架无缝集成。
易于配置: 相对简单，适合中小型项目。
灵活的消息模式: 支持点对点（Queue）和发布/订阅（Topic）两种模式。

缺点

性能较低: 相比 RabbitMQ 和 Kafka，在高吞吐量场景下性能较差。
缺乏现代特性: 不如 Kafka 新颖，缺少一些现代流处理特性。
社区活跃度下降: 相比其他两个工具，社区活跃度有所下降。
复杂性: 在大型系统中配置和管理相对复杂。

总结

特征	RabbitMQ	Apache Kafka	ActiveMQ
成熟稳定性	高	高	高
协议支持	多种协议 (AMQP, MQTT, STOMP)	主要支持 Kafka 协议	多种协议 (TCP, UDP, HTTP)
可靠性	强	极强	强
易于管理	是，提供强大的管理界面	较复杂，需要配置 Zookeeper	是，易于配置
性能	中等	高	中等
扩展性	中等	非常好，支持水平扩展	中等
持久化	强	极强	强
容错性	强	极强	强
流处理能力	基本支持	内置流处理功能	基本支持
学习曲线	中等	陡峭	中等
资源消耗	中等	高	中等

选择哪种消息队列工具取决于你的具体需求，例如性能要求、可靠性需求、现有技术栈以及团队的技术背景。

以上内容涵盖了如何在 Java 微服务中使用 RabbitMQ、Apache Kafka 和 ActiveMQ 进行消息队列处理。每个部分都包含了安装步骤、基本概念和示例代码。你可以根据实际需求选择合适的工具，并进一步扩展和完善这些示例。

原文地址：https://blog.csdn.net/xiaozukun/article/details/145251814

免责声明：本站文章内容转载自网络资源，如侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：c# PDF文件合并工具
下一篇：iOS UIScrollView的一个特性

大华大数据开发面试题及参考答案
内存泄露是指程序在运行过程中，不断申请内存空间却没有及时释放不再使用的内存，导致可用内存逐渐减少，最终可能影响程序的正常运行甚至导致系统崩溃的现象。对象引用未正确释放：当对象之间存在强引用关系且不再需
阅读更多2025-01-21
对羟基苯甲酸行业分析
这些企业不仅拥有先进的生产工艺和技术，还通过持续的技术创新和市场拓展，巩固了其在全球市场的领先地位。2021年，全球前四大厂商占有大约83.0%的市场份额。此外，对羟基苯甲酸的水溶性虽低，但在乙醇、乙
阅读更多2025-01-21
麦田物语学习笔记:制作[SceneName]Attribute特性
因为在现有的项目中,像开始场景的切换或者Telepot组件都需要手动输入场景名,有时还可能键入出错,而该特性能用选择的方式去解决这一问题。
阅读更多2025-01-21
MySQL主从同步
MySQL 主从同步是一种数据库复制技术，它通过将主服务器上的数据更改复制到一个或多个从服务器，实现数据的自动同步。主从同步的核心原理是将主服务器上的二进制日志复制到从服务器，并在从服务器上执行这些日
阅读更多2025-01-21
Spring Boot整合WebSocket
WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议。它允许客户端和服务器之间进行实时数据交换，而无需每次都重新建立连接。WebSocket 的核心特性包括：双向通信：客户端和服务器都
阅读更多2025-01-21
Javascript IndexedDB 数据库
IndexedDB是一个强大的客户端存储解决方案，适合存储大量结构化数据，支持异步操作和查询。它比更加灵活和高效，但相对而言也更为复杂。适用于需要存储大量数据或进行复杂查询的场景。如果你的应用需要存储
阅读更多2025-01-21
RK3568笔记七十六：使用V4L2框架录制MP4视频保存到本地
若该文为原创文章，转载请注明原文出处。录制MP4使用的是ffmpeg，如何编译自行处理。
阅读更多2025-01-21
比赛的冠亚季军
比赛的冠亚季军
阅读更多2025-01-21
数据库基础知识：记录、表、字段、数据类型、约束、主键、外键、规范化、索引、序列
数据库是由一个或多个有组织的数据集合组成，而数据库管理系统（）是操作数据库的软件，包括很多人本科上课学过的SQL Server，现在常用的MySQL、Postgresql等，用于提供数据的存储、访问、
阅读更多2025-01-21
基于Python的多元医疗知识图谱构建与应用研究（下）
在构建基于医疗知识图谱的医疗知识图谱数据层时，数据源的选择与获取至关重要。数据源的质量和丰富度直接决定了知识图谱的可靠性和实用性。医学文献是重要的数据源之一，包括学术期刊论文、医学研究报告等。这些文献
阅读更多2025-01-21

java微服务中消息队列处理中间件基础语法学习，零基础学习

1. RabbitMQ

1.1 安装 RabbitMQ

1.2 基本概念

1.3 示例代码

1.3.1 创建 Spring Boot 项目

1.3.2 配置 RabbitMQ

1.3.3 生产者

1.3.4 消费者

1.3.5 控制器

2. Apache Kafka

2.1 安装 Kafka

2.2 基本概念

2.3 示例代码

2.3.1 创建 Spring Boot 项目

2.3.2 配置 Kafka

2.3.3 创建 Topic

2.3.4 生产者

2.3.5 消费者

2.3.6 控制器

3. ActiveMQ

3.1 安装 ActiveMQ

3.2 基本概念

3.3 示例代码

3.3.1 创建 Spring Boot 项目

3.3.2 配置 ActiveMQ

3.3.3 生产者

3.3.4 消费者

3.3.5 控制器

4.RabbitMQ、Apache Kafka 和 ActiveMQ 在微服务架构中作为消息队列处理工具的优缺点对比。

1. RabbitMQ

优点

缺点

2. Apache Kafka

优点

缺点

3. ActiveMQ

优点

缺点

总结

相关文章