本资源提供了一个使用Java编写的Kafka生产消费程序的示例，旨在帮助理解Kafka在Java环境下的工作方式。Kafka是一个高吞吐量的消息系统，主要用Scala编写，其生产和消费机制与传统消息系统有所区别。
在开始之前，你需要创建一个Maven项目，并在项目的`pom.xml`文件中添加Kafka的相关依赖。这里引用的Kafka版本为0.8.0，对应的依赖如下：
```xml
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<version>0.8.0</version>
</dependency>
```
接下来，我们将详细讲解如何编写Java生产者和消费者代码。
1. Kafka生产者（Producer）
在Java中，生产者用于发送消息到Kafka的主题（Topic）。以下是一个简单的`KafkaProducer`类示例：
```java
package cn.outofmemory.kafka;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public class KafkaProducer {
private final Producer<String, String> producer;
public static final String TOPIC = "TEST-TOPIC";
public KafkaProducer() {
Properties props = new Properties();
// 配置Kafka服务器地址
props.put("metadata.broker.list", "192.168.193.148:9092");
// 配置value和key的序列化类
props.put("serializer.class", "kafka.serializer.StringEncoder");
// 配置请求确认模式
props.put("request.required.acks", "1");
// 创建Producer实例
producer = new Producer<>(new ProducerConfig(props));
}
// 发送消息到指定主题
public void sendMessage(String key, String message) {
producer.send(new KeyedMessage<>(TOPIC, key, message));
}
// 关闭生产者
public void close() {
producer.close();
}
}
```
在这个示例中，我们创建了一个`KafkaProducer`类，初始化时设置Kafka服务器地址、序列化类以及请求确认模式。`sendMessage`方法用于发送消息，`close`方法用于关闭生产者连接。
2. Kafka消费者（Consumer）
消费者则从Kafka的主题中接收并处理消息。这里我们创建一个`KafkaConsumer`类：
```java
package cn.outofmemory.kafka;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
public class KafkaConsumer {
private final ConsumerConnector consumer;
public KafkaConsumer(String zookeeper, String group) {
Properties props = new Properties();
props.put("zookeeper.connect", zookeeper);
props.put("group.id", group);
props.put("zookeeper.session.timeout.ms", "5000");
props.put("zookeeper.sync.time.ms", "2000");
props.put("auto.commit.interval.ms", "1000");
consumer = kafka.consumer.Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
}
// 获取消息流
public Map<String, List<KafkaStream<byte[], byte[]>>> getMsgStreams(int numThreads) {
Map<String, Integer> topicCountMap = new HashMap<>();
topicCountMap.put(KafkaProducer.TOPIC, numThreads);
Map<String, List<KafkaStream<byte[], byte[]>>> streams = consumer.createMessageStreams(topicCountMap);
return streams;
}
// 开始消费
public void startConsuming(List<KafkaStream<byte[], byte[]>> streams) {
for (final KafkaStream<byte[], byte[]> stream : streams) {
new Thread(new Runnable() {
@Override
public void run() {
ConsumerIterator<byte[], byte[]> it = stream.iterator();
while (it.hasNext()) {
System.out.println(new String(it.next().message()));
}
}
}).start();
}
}
// 关闭消费者
public void shutdown() {
consumer.shutdown();
}
}
```
`KafkaConsumer`类创建了消费者实例，通过`getMsgStreams`获取消息流，然后在`startConsuming`中启动多线程进行消息消费。
在实际应用中，你可能需要根据自己的需求对这些示例代码进行调整，例如设置不同的序列化类、调整消费者分组策略、处理错误等。同时，确保Kafka服务器已正确安装和配置，Zookeeper也在运行，以便生产者和消费者能够正常通信。
我的内容管理
展开
