项目场景:
互联网项目中经常用到MQ,由于本地项目开发连接测试环境kafka很不方便,所有在本机搭建一个kafka,方便开发测试。
前置准备
提示:Kafka的运行依赖于Zookeeper,所以在运行Kafka之前我们需要安装并运行Zookeeper
下载Zookeeper地址:https://zookeeper.apache.org/releases.html
下载kafka地址:http://kafka.apache.org/downloads.html
配置Zookeeper
1.将下载好的文件解压到本地,如图:
复制zoo_sample.cfg文件,并将新复制的文件命名为zoo.cfg,修改文件zoo.cfg内容如下:
dataDir=F:\mq\apache-zookeeper-3.6.3\dataDir
dataLogDir=F:\mq\apache-zookeeper-3.6.3\dataLogDir
2.配置Window环境变量
3.启动Zookeeper
进入Zookeeper安装目录,cmd 输入命令zkserver,如图
启动成功!!
配置kafka
1.解压下载文件到本地
进入F:\mq\kafka_2.13-2.8.0\config文件内,修改文件server.properties
log.dirs=F:\mq\kafka_2.13-2.8.0\logs
2.启动kafka服务
在安装目录cmd输入命令:
.\bin\windows\kafka-server-start.bat .\config\server.properties
无报错则正常启动,本地启动窗口不要关闭。
3.创建topic名称为syn_user的命令:
.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic syn_user
4.查看创建的topic
.\bin\windows\kafka-topics.bat --list --zookeeper localhost:2181
本地kafka环境测试:
启动生产者
.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic syn_user
启动消防者监听消息文章来源:https://www.toymoban.com/news/detail-428772.html
.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic syn_user
文章来源地址https://www.toymoban.com/news/detail-428772.html
springboot 集成:
1.引入pom依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
2.yml配置
kafka:
bootstrap-servers: 127.0.0.1:9092
producer:
acks: -1
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
consumer:
enable-auto-commit: false
key-serializer: org.apache.kafka.common.serialization.StringDeserializer
value-serializer: org.springframework.kafka.support.serializer.JsonDeserializer
group-id: test-consumer-group
listener:
ack-mode: MANUAL
3.创建消息生产者
@RestController
@Api(value = "mq消息", tags = "Fh-mq消息")
@RequestMapping("/wkafka")
public class ProducerController {
private static final String KAFKA_TOPIC_NAME = "wlhydemo";
@Autowired
KafkaTemplate<String, String> kafka;
@PostMapping("/send")
public String register(@RequestBody User user) {
try {
String message = JSONUtil.toJsonStr(user);
System.out.println("注册用户信息:" + message);
kafka.send(KAFKA_TOPIC_NAME, message);
return "OK";
} catch (Exception e) {
e.printStackTrace();
}
return "消息同步失败";
}
}
4.监听topic消息类
@Slf4j
@Component
public class KaUserConsumer {
@KafkaListener(topics = "wlhydemo")
public void listenFlowStart(@Payload String businessStr,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
@Header(KafkaHeaders.OFFSET) int offset )
{
try{
// 模拟业务处理...
log.info("当前消费分区:{}", partition);
log.info("当前消费位置:{}", offset);
log.info("接收到的消息:{}", businessStr);
User user= JSONUtil.toBean(businessStr, User.class);
user.getNickName();
} catch (Exception e) {
e.printStackTrace();
}
}
}
到了这里,关于Window下搭建kafka运行环境的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!