直接上干货，我们公司最近要进行消息推送指定软件kafka,直接走起。
1.下载
kafka 是apache的项目。下载地址：kafka.apache.org/
点击download kafka 进入查看相关版本进行下载。
我这里用的版本比窘旧一点，公司技术一切求稳。
下载好安装包就已经实现了。
2.安装
说安装就是免安装版本，简单配置一下就可以使用了
直接解压就可以，但是需要我们修改一些配置文件。
kafka 集成了 zookeeper 的软件包，不需要安装，配置就好。
1）：配置zookeeper.properties
进入到config文件夹里面，找到zookeeper.properties文件，进行编辑，找到dataDir，修改这个参数的路径为：
dataDir=D:\kafka\kafka_2.13-3.2.1\zookeeper-data (安装目录\zookeeper-data)
2）：配置server.properties
进入到config文件夹里面，找到server.properties文件，进行编辑，找到log.dirs，修改这个参数的路径为：
log.dirs=D:\kafka\kafka_2.13-3.2.1\kafka-logs (安装目录\kefke-log)
完成了配置，现在就可以启动了
3.启动
1）：启动zookeeper
打开cmd，进入kafka安装目录，执行：
.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties2）：启动kafka
再打开一个cmd，进入kafka安装目录，执行：
.\bin\windows\kafka-server-start.bat .\config\server.properties完成启动就可以访问了，但是我们访问有端口限制，所以我又进行了端口修改
kafka默认端口号：9092 我这里给它改为8081 方便测试访问。
4.配置端口（可以不修改）
config下
1、service.properties (添加) port = 8081 不指定的话，按照默认9092
2、connect-distributed.properties (集群) bootstrap.servers=localhost:8081
3、producer.properties (生产者) bootstrap.servers=localhost:8081
4、connect-standalone.properties (单机) bootstrap.servers=localhost:8081
5、consumer.properties (消费者) bootstrap.servers=localhost:8081
根据情况配置端口
5.编码
maven坐标
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.2.1</version>
</dependency>1）生产者代码
package com.java;
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.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
public class MyKafKaProducer extends Thread{
KafkaProducer<Integer, String> producer;
String topic; // 主题
public MyKafKaProducer(String topic) {
// 构建连接配置
Properties properties = new Properties();
// 若要配多个服务器，用逗号隔开
// 注：服务器要开放端口，若云服务器还要在server.properties配置内网IP和外网IP
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.32:8081");
properties.put(ProducerConfig.CLIENT_ID_CONFIG, "my-producer");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
IntegerSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
// 构造Client无非是：new 或 工厂模式
producer = new KafkaProducer<Integer, String>(properties);
this.topic = topic;
}
public void run() {
int num = 0;
String msg = "kafka practice msg: " + num;
while (num < 20) {
try {
// 发送消息send()！！! 同步调用
// Future.get()会阻塞，等待返回结果......
RecordMetadata recordMetadata = producer.send(new ProducerRecord<>(topic, msg)).get();
// 等上面get到结果了，才能执行这里
System.out.println(recordMetadata.offset() + "->" + recordMetadata.partition() +
"->" + recordMetadata.topic());
TimeUnit.SECONDS.sleep(2);
num++;
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
// 传入test主题
new MyKafKaProducer("test").start();
}
}
2）消费者代码
package com.java;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class MykafkaConsumer extends Thread{
KafkaConsumer<Integer, String> consumer;
String topic;
public MykafkaConsumer(String topic) {
// 构建连接配置，这里是ConsumerConfig
Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.32:8081");
properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "my-consumer");
// 反序列化，这里是Deserializer
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
IntegerDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
// 以下是Producer没有的配置
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-gid"); // 要加入的group
properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); // 超时，心跳
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 自动提交（批量）
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 新group消费为位置
consumer = new KafkaConsumer<>(properties);
this.topic = topic;
}
public void run() {
// 死循环不断消费消息
while (true) {
// 绑定订阅主题
// 注：Collections.singleton返回一个单元素&不可修改Set集合，
// 同样的还有singletonList，singletonMap
consumer.subscribe(Collections.singleton(this.topic));
// 接收消息 POLL()！！！
ConsumerRecords<Integer, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
// 注：一行的lamada表达式可以不用{}
consumerRecords.forEach(record -> System.out.println(record.key() + "->" +
record.value() + "->" + record.offset()));
}
}
public static void main(String[] args) {
// 拉取test主题的消息
new MykafkaConsumer("test").start();
}
}
topic 主题必须一致才可以实现生产者和消费者的互通
