kafka单机环境搭建以及springboot整合
kafka单机环境搭建虚拟机版本是 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 -zxvfxxxxxxx(tar包名称)修改配置文件cd /usr
·
kafka单机环境搭建
虚拟机版本是 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
- 首先导入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- kafka已经被springoot-starer整合 因此不需要带版本号即可 -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.15</version>
</dependency>
- 编写配置
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;
/**
* @Author llf
* @Date 2021/12/13 15:05
* @Version 1.0
*/
@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;
/**
* @Author llf
* @Date 2021/12/13 15:07
* @Version 1.0
*
* 创建生产者
*/
@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;
/**
* @Author llf
* @Date 2021/12/13 15:16
* @Version 1.0
*
* 创建消费者
*/
@Component
@Slf4j
public class UserLogConsumer {
/**
* 指定topic 进行消费
* @param consumerRecord
*/
@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);
}
}
将服务器的防火墙关闭 启动主启动类进行测试
创建消费者查看信息
更多推荐
已为社区贡献1条内容
所有评论(0)