java八股文之消息中间件

embedded/2025/3/17 7:03:05/

<在Java中使用消息中间件时,通常会选择一些流行的开源解决方案,如Apache Kafka、RabbitMQ、ActiveMQ等。这些消息中间件提供了高效、可靠的消息传递机制,广泛应用于企业级应用中。下面我将介绍如何在Java中使用Apache Kafka进行消息传递的基本步骤和示例代码。

1. 添加Kafka依赖

首先,你需要在你的Java项目中添加Kafka的依赖。如果你使用Maven,可以在pom.xml文件中添加以下依赖:

<dependencies>

    <!-- Apache Kafka clients -->

    <dependency>

        <groupId>org.apache.kafka</groupId>

        <artifactId>kafka-clients</artifactId>

        <version>3.0.0</version> <!-- 请使用最新版本 -->

    </dependency>

</dependencies>

2. 创建生产者

生产者(Producer)是发送消息到Kafka的一个客户端。以下是一个简单的生产者示例:

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.ProducerRecord;

import org.apache.kafka.clients.producer.ProducerConfig;

import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class SimpleProducer {

    public static void main(String[] args) {

        // 配置生产者属性

        Properties props = new Properties();

        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // Kafka服务器地址

        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 创建生产者实例

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 发送消息

        String topic = "test-topic";

        String value = "Hello, Kafka!";

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, value);

        producer.send(record);

        System.out.println("Message sent successfully!");

        // 关闭生产者

        producer.close();

    }

}

3. 创建消费者

消费者(Consumer)是从Kafka接收消息的一个客户端。以下是一个简单的消费者示例:

import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.apache.kafka.clients.consumer.ConsumerRecords;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;

import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;

import java.util.Collections;

import java.util.Properties;

public class SimpleConsumer {

    public static void main(String[] args) {

        // 配置消费者属性

        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // Kafka服务器地址

        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group"); // 消费者组ID

        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早的记录开始消费

        // 创建消费者实例

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Collections.singletonList("test-topic")); // 订阅主题

        // 轮询消息并处理它们

        while (true) {

            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); // 轮询消息,等待时间为100ms

            for (ConsumerRecord<String, String> record : records) {

                System.out.printf("Received message: (%s, %s)%n", record.key(), record.value());

            }

        }

    }

}

4. 运行和测试

确保你的Kafka服务正在运行,并且主题test-topic已经创建。你可以使用Kafka自带的命令行工具kafka-topics来创建主题:

kafka-topics --create --topic test-topic


http://www.ppmy.cn/embedded/173278.html

相关文章

58.Harmonyos NEXT 图片预览组件架构设计与实现原理

温馨提示&#xff1a;本篇博客的详细代码已发布到 git : https://gitcode.com/nutpi/HarmonyosNext 可以下载运行哦&#xff01; Harmonyos NEXT 图片预览组件架构设计与实现原理 文章目录 Harmonyos NEXT 图片预览组件架构设计与实现原理效果预览一、组件架构概述1. 核心组件层…

2.12[A]distribute sys

在分布式训练中&#xff0c;特别是使用3D并行&#xff08;数据并行、流水线并行和模型并行&#xff09;时&#xff0c;不同阶段的GPU可能因为通信或数据依赖而出现空闲时间&#xff0c;这些空闲时间就是所谓的“气泡”。这些气泡会降低整体的训练效率&#xff0c;导致GPU资源的…

<rust><tauri><GUI>基于tauri和rust,编写一个二维码生成器

前言 本文是基于rust和tauri&#xff0c;由于tauri是前、后端结合的GUI框架&#xff0c;既可以直接生成包含前端代码的文件&#xff0c;也可以在已有的前端项目上集成tauri框架&#xff0c;将前端页面化为桌面GUI。 环境配置 系统&#xff1a;windows 10平台&#xff1a;vis…

【机器学习-基础知识】统计和贝叶斯推断

1. 概率论基本概念回顾 1. 概率分布 定义: 概率分布(Probability Distribution)指的是随机变量所有可能取值及其对应概率的集合。它描述了一个随机变量可能取的所有值以及每个值被取到的概率。 对于离散型随机变量,使用概率质量函数来描述。对于连续型随机变量,使用概率…

HOT100系列——(普通数组+矩阵)

普通数组 1.合并区间 56. 合并区间https://leetcode.cn/problems/merge-intervals/ 先针对左区间进行排序&#xff0c;这样可以对右边进行考虑&#xff0c;如果intervals 中一个新的区间的左端点小于原ans的右端点&#xff0c;那么就能合并&#xff0c;右端点合并成原p【1】和…

基于SpringBoot3+Druid数据库连接池与外部PostgreSQL的Kubernetes Pod YAML全解析

说明 一个基于Spring Boot 3 Druid 外部PostgreSQL的Kubernetes Pod YAML详细解析&#xff0c;包含最佳实践和关键配置说明&#xff1a; YAML apiVersion: apps/v1 kind: Deployment metadata:name: springboot-applabels:app: springboot-app spec:replicas: 2selector:ma…

Spring AI 与 LangChain4j 选型对比分析

Spring AI 与 LangChain4j 选型对比分析 对比分析目前市场上两个主流的Java开源AI框架——Spring AI 和 LangChain4j&#xff0c;探讨它们的技术特点、适用场景以及未来发展的潜力。 一、易用性对比 Spring AI 优点 &#xff1a;对 Spring 开发者友好&#xff0c;通过自动配置…

Jdk7不同集合的扩容机制

Jdk7不同集合的扩容机制 集合类初始容量负载因子扩容公式扩容时机ArrayList10无新容量 旧容量 1.5元素数量超过容量时HashMap160.75新容量 旧容量 2元素数量超过 容量 负载因子 时HashSet160.75新容量 旧容量 2元素数量超过 容量 负载因子 时Vector10无新容量 旧容量…