java八股文之消息中间件

<在Java中使用消息中间件时，通常会选择一些流行的开源解决方案，如Apache Kafka、RabbitMQ、ActiveMQ等。这些消息中间件提供了高效、可靠的消息传递机制，广泛应用于企业级应用中。下面我将介绍如何在Java中使用Apache Kafka进行消息传递的基本步骤和示例代码。 1 添加Kafka依赖 首先，你需要在你的Java项目中添加Kafka的依赖。如果你使用Maven，可以在pom.xml文件中添加以下依赖：

org.apache.kafka kafka-clients 3.0.0 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 producer = new KafkaProducer<>(props); // 发送消息 String topic = “test-topic”; String value = “Hello, Kafka!”; ProducerRecord 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 consumer = new KafkaConsumer<>(props); consumer.subscribe(Collections.singletonList(“test-topic”)); // 订阅主题 // 轮询消息并处理它们 while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); // 轮询消息，等待时间为100ms for (ConsumerRecord 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