虚拟机版本是 centos7 64位
tar包下载进入到cd /usr/local目录下 使用wget命令从官网下载压缩包文件 下载bin文件 不要下载源码文件
wget https://dlcdn.apache.org/kafka/3.0.0/kafka_2.12-3.0.0.tgz 这个是2.12版本的解压
tar -zxvf xxxxxxx（tar包名称）修改配置文件
cd /usr/local/kafka_2.12-3.0.0/config
- 修改server.properties
vim server.properties 修改文件保存位置 默认的/tmp/xxx 路径下的文件 在服务器重启后会数据丢失 log.dirs=/opt/kafka/log 配置监听器 时刻监听端口信息 使消费者可以第一时间拿到生产者生产的信息 listeners=PLAINTEXT://192.168.171.224:9092 advertised.listeners=PLAINTEXT://192.168.171.224:9092 其他使用默认配置 192.168.171.224 是端口号 改成自己的端口号即可
- 修改 zookeeper.properties
vim zookeeper.properties 修改的是kafka内置的zookeeper配置 如果启动服务时使用的不是内置的zookeeper 就不需要改 dataDir=/opt/zookeeper/data 其余默认配置即可启动服务
-
首先启动zookeeper服务 使用 -daemon守护进程命令 让服务在后台运行
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
-
然后启动kafka服务 同理使用-daemon守护进程命令 让服务在后台运行
bin/kafka-server-start.sh -daemon config/server.properties
- 启动成功后 通过lsof -i:port命令 查看服务是否正常启动
psof -i:9092 kafka默认端口 psof -i:2181 zookeeper默认端口
- 确定无误后 开始创建主题
bin/kafka-topics.sh --create --bootstrap-server 192.168.171.224:9092 --replication-factor 1 --partitions 1 --topic test2 创建名称为test2的主题 192.168.171.224 这个为服务器端口号 换成自己的即可
- 主题创建完成 开始创建生产者
bin/kafka-console-producer.sh --broker-list 192.168.171.224:9092 --topic test 命令运行完毕 会呈现出输入状态 > 编辑需要发布的内容 回车键切换下一条信息编辑 >
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-15EokYcE-1639386732241)(E:/Typroa图片/image-20211213145000951.png)]
- 创建消费者
bin/kafka-console-consumer.sh --bootstrap-server 192.168.171.224:9092 --topic test --from-beginning 命令执行成功 会直接显示生产者发布的信息内容 并一直监听 只要发布新的消息 就会直接显示
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lQSIxgZe-1639386732242)(E:/Typroa图片/image-20211213145144223.png)]
至此 单机版kafka环境算搭建成功
关闭服务bin/kafka-server-stop.sh config/server.properties 关闭kafka进程 bin/zookeeper-server-stop.sh config/zookeeper.properties 关闭zookeeper进程kafka整合springboot
- 首先导入依赖
org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-test test org.springframework.kafka spring-kafka org.projectlombok lombok com.alibaba fastjson 1.2.15
- 编写配置
spring.application.name=kafka-user server.port=8080 #============== kafka =================== # 指定kafka 代理地址，可以多个 spring.kafka.bootstrap-servers=192.168.171.224:9092 #=============== provider ======================= spring.kafka.producer.retries=0 # 每次批量发送消息的数量 spring.kafka.producer.batch-size=16384 spring.kafka.producer.buffer-memory=33554432 # 指定消息key和消息体的编解码方式 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer #=============== consumer ======================= # 指定默认消费者group id spring.kafka.consumer.group-id=user-log-group spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.enable-auto-commit=true spring.kafka.consumer.auto-commit-interval=100 # 指定消息key和消息体的编解码方式 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer创建消息实体类
package com.llf.pojo;
import lombok.Data;
import lombok.experimental.Accessors;
@Accessors(chain = true)
@Data
public class UserLogs {
private String username;
private String userId;
private String state;
}
创建生产者
package com.llf.producer;
import com.alibaba.fastjson.JSON;
import com.llf.pojo.UserLogs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
@Component
public class UserLogProducer {
@Autowired
private KafkaTemplate kafkaTemplate;
public void sendLog(String userid){
UserLogs userLog = new UserLogs();
userLog.setUsername("jhp").setUserId(userid).setState("0");
System.err.println("发送用户日志数据:"+userLog);
kafkaTemplate.send("test", JSON.toJSONString(userLog));
}
}
创建消费者
package com.llf.consumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.Optional;
@Component
@Slf4j
public class UserLogConsumer {
@KafkaListener(topics = {"test"})
public void consumer(ConsumerRecord consumerRecord){
//判断是否为null
Optional kafkaMessage = Optional.ofNullable(consumerRecord.value());
log.info(">>>>>>>>>> record =" + kafkaMessage);
if(kafkaMessage.isPresent()){
//得到Optional实例中的值
Object message = kafkaMessage.get();
System.err.println("消费消息:"+message);
}
}
}
在主启动类引用生产者方法
package com.llf;
import com.llf.producer.UserLogProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import javax.annotation.PostConstruct;
@SpringBootApplication
public class KafkaApplication {
@Autowired
private UserLogProducer kafkaSender;
@PostConstruct
public void init(){
for (int i = 0; i < 10; i++) {
//调用消息发送类中的消息发送方法
kafkaSender.sendLog(String.valueOf(i));
}
}
public static void main(String[] args) {
SpringApplication.run(KafkaApplication.class, args);
}
}
将服务器的防火墙关闭 启动主启动类进行测试
创建消费者查看信息
