简易 Kafka 消息队列系统（Java）

1️⃣ 核心思路

Topic（主题）：用来存储消息的队列。

Broker：管理多个 Topic，负责接收消息和分发给消费者。

Producer（生产者）：发送消息到指定 Topic。

Consumer（消费者）：从 Topic 拉取消息消费。

多线程安全：使用 BlockingQueue 来保证消息队列的线程安全。

2️⃣ Broker 实现

import java.util.concurrent.*;

import java.util.*;

public class SimpleBroker {

private Map<String, BlockingQueue<String>> topics = new ConcurrentHashMap<>();

// 创建主题

public void createTopic(String topicName) {

topics.putIfAbsent(topicName, new LinkedBlockingQueue<>());

System.out.println("Topic created: " + topicName);

}

// 发送消息

public void send(String topicName, String message) {

BlockingQueue<String> queue = topics.get(topicName);

if (queue != null) {

queue.offer(message);

System.out.println("Message sent to " + topicName + ": " + message);

} else {

System.out.println("Topic does not exist: " + topicName);

}

// 拉取消息

public String poll(String topicName) throws InterruptedException {

BlockingQueue<String> queue = topics.get(topicName);

if (queue != null) {

return queue.take(); // 阻塞直到有消息

}

return null;

}

3️⃣ Producer 实现

public class SimpleProducer {

private SimpleBroker broker;

public SimpleProducer(SimpleBroker broker) {

this.broker = broker;

}

public void send(String topic, String message) {

broker.send(topic, message);

}

4️⃣ Consumer 实现

public class SimpleConsumer implements Runnable {

private SimpleBroker broker;

private String topic;

public SimpleConsumer(SimpleBroker broker, String topic) {

this.broker = broker;

this.topic = topic;

}

@Override

public void run() {

try {

while (true) {

String msg = broker.poll(topic);

System.out.println(Thread.currentThread().getName() + " consumed from " + topic + ": " + msg);

}

} catch (InterruptedException e) {

System.out.println(Thread.currentThread().getName() + " stopped.");

}

5️⃣ 测试系统

public class Main {

public static void main(String[] args) throws InterruptedException {

SimpleBroker broker = new SimpleBroker();

broker.createTopic("test-topic");

SimpleProducer producer = new SimpleProducer(broker);

// 创建两个消费者

Thread consumer1 = new Thread(new SimpleConsumer(broker, "test-topic"), "Consumer-1");

Thread consumer2 = new Thread(new SimpleConsumer(broker, "test-topic"), "Consumer-2");

consumer1.start();

consumer2.start();

// 生产消息

for (int i = 1; i <= 10; i++) {

producer.send("test-topic", "message-" + i);

Thread.sleep(500); // 模拟生产速度

}

// 等待几秒

Thread.sleep(5000);

// 停止消费者

consumer1.interrupt();

consumer2.interrupt();

}

6️⃣ 功能说明

主题管理

可以创建多个 Topic，每个 Topic 有自己的消息队列。

线程安全

使用 BlockingQueue，支持多生产者、多消费者。

消费者阻塞等待消息

take() 方法会在队列为空时阻塞，类似 Kafka 消费者拉取机制。

简单分发

每个消费者都可以独立消费消息。

如果需要 Kafka 式的消费者组，可以在每个 Topic 中维护消费者偏移量（可以扩展）。

ITPub博客

java实现简易 Kafka 消息队列系统。

最新文章