download

https://kafka.apache.org/downloads

这里用的版本是3.1.0，scala版本为2.12

启动命令

https://kafka.apache.org/quickstart

1. 启动zookeeper

bin/zookeeper-server-start.sh config/zookeeper.properties

2. 启动kafka

bin/kafka-server-start.sh config/server.properties

3. 创建topic

quickstart-events是topic的名称

bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

4. 查看topic信息

bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092

5. 写入数据

新开一个bash

bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092

6. 读数据

新开一个bash

bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092

Java连接kafka

参考这篇文章，但是配置代码改了一下

1. 引入maven依赖

        
            org.apache.kafka
            kafka-log4j-appender
            0.10.0.0
        
        
            org.apache.kafka
            kafka_2.12
            1.0.0
            provided
        

        
            org.apache.kafka
            kafka-clients
            1.0.0
        

2. Producer

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

public class KafkaProducerTest implements Runnable {

    private final KafkaProducer producer;
    private final String topic;
    public KafkaProducerTest(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        this.producer = new KafkaProducer(props);
        this.topic = topicName;
    }

    @Override
    public void run() {
        int messageNo = 1;
        System.out.println("1");
        try {
            for(;;) {
                String messageStr="你好，这是第"+messageNo+"条数据";
                producer.send(new ProducerRecord(topic, "Message", messageStr));
                //生产了100条就打印
                if(messageNo%100==0){
                    System.out.println("发送的信息:" + messageStr);
                }
                //生产1000条就退出
                if(messageNo%1000==0){
                    System.out.println("成功发送了"+messageNo+"条");
                    break;
                }
                messageNo++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

    public static void main(String args[]) {
        KafkaProducerTest test = new KafkaProducerTest("quickstart-events");
        Thread thread = new Thread(test);
        thread.start();
    }
}

3. Consumer

import java.util.Arrays;
import java.util.Properties;
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;

public class kafkaConsumerTest implements Runnable {

	private final KafkaConsumer consumer;
	private ConsumerRecords msgList;
	private final String topic;
	private static final String GROUPID = "groupB";

	public kafkaConsumerTest(String topicName) {
		Properties props = new Properties();
		props.put("bootstrap.servers", "localhost:9092");
		props.put("group.id", GROUPID);
		props.put("enable.auto.commit", "true");
		props.put("auto.commit.interval.ms", "1000");
		props.put("session.timeout.ms", "30000");
		props.put("auto.offset.reset", "earliest");
		props.put("key.deserializer", StringDeserializer.class.getName());
		props.put("value.deserializer", StringDeserializer.class.getName());
		this.consumer = new KafkaConsumer(props);
		this.topic = topicName;
		this.consumer.subscribe(Arrays.asList(topic));
	}

	@Override
	public void run() {
		int messageNo = 1;
		System.out.println("---------开始消费---------");
		try {
			for (;;) {
					msgList = consumer.poll(1000);
					if(null!=msgList&&msgList.count()>0){
					for (ConsumerRecord record : msgList) {
						//消费100条就打印 ,但打印的数据不一定是这个规律的
						if(messageNo%100==0){
							System.out.println(messageNo+"=======receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset());
						}
						//当消费了1000条就退出
						if(messageNo%1000==0){
							break;
						}
						messageNo++;
					}
				}else{	
					Thread.sleep(1000);
				}
			}		
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			consumer.close();
		}
	}  
	public static void main(String args[]) {
		kafkaConsumerTest test1 = new kafkaConsumerTest("quickstart-events");
		Thread thread1 = new Thread(test1);
		thread1.start();
	}
}