首先需要先引入pom.xml依赖。

<dependency>
    <groupId>org.apache.kafka</groupId>
	<artifactId>kafka-clients</artifactId>
	<version>0.10.2.0</version>
</dependency>

引入依赖后，实现生产者和消费者代码
生产者

package com.mall.kafka.KafkaDemo;
import java.util.Properties;
import java.util.Random;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
/**
 * 生产者
 * 
 * @author AdminMall
 *
 */
public class Producer {
	public static String topic = "malltest";// 定义主题
	public static void main(String[] args) throws InterruptedException {
		Properties p = new Properties();
		// kafka地址，多个地址用逗号分割
		p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
		// 配置value的序列化类
		p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		// 配置key的序列化类
		p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		// 0表示不确认主服务器是否收到消息,马上返回,低延迟但最弱的持久性,数据可能会丢失
		// 1表示确认主服务器收到消息后才返回,持久性稍强,可是如果主服务器死掉,从服务器数据尚未同步,数据可能会丢失
		// -1表示确认所有服务器都收到数据,完美!
		p.put("request.required.acks", "-1");
		// 异步生产,批量存入缓存后再发到服务器去
		p.put("producer.type", "async");
		KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(p);
		try {
			ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, "测试消息");
			kafkaProducer.send(record);
			System.out.println("推送成功");
		} finally {
			// 必须关闭
			kafkaProducer.close();
		}
	}
}

消费者

package com.mall.kafka.KafkaDemo;
import java.util.Collections;
import java.util.Properties;
import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;
/**
 * 消费者
 * 
 * @author AdminMall
 *
 */
public class Consumer {
	public static void main(String[] args) {
		Properties p = new Properties();
		p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
		p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		// 自动提交秒
		// p.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, false);
		// 是否开启自动提交
		p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
		// group 代表一个消费组,加入组里面,消息只能被该组的一个消费者消费
		// 如果所有消费者在一个组内,就是传统的队列模式,排队拿消息
		// 如果所有的消费者都不在同一个组内,就是发布-订阅模式,消息广播给所有组
		// 如果介于两者之间,那么广播的消息在组内也是要排队的
		p.put(ConsumerConfig.GROUP_ID_CONFIG, "malltest");
		KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(p);
		// 订阅消息；subscribe()
		// 方法接受一个主题列表作为参数：也可以接收一个正则表达式，匹配多个主题；consumer.subscribe("test.*");
		kafkaConsumer.subscribe(Collections.singletonList(Producer.topic));
		while (true) {
			// 100 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
			ConsumerRecords<String, String> records = kafkaConsumer.poll(100);

			// poll 返回一个记录列表。
			// 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。
			for (ConsumerRecord<String, String> record : records) {
				// 主题
				System.out.println("主题：" + record.topic());
				System.out.println("读取位置：" + record.offset());
				System.out.println("Key：" + record.key());
				System.out.println("内容：" + record.value());
			}
			/ 手动提交
			try {
				// poll 的数据全部处理完提交
				kafkaConsumer.commitAsync();
			} catch (CommitFailedException e) {
				e.printStackTrace();
			}
		}
	}
}