openwrt golang mysql_windows下golang实现Kfaka消息发送及kafka环境搭建 - Go语言中文社区

openwrt golang mysql_windows下golang实现Kfaka消息发送及kafka环境搭建


kafka环境搭建:

一、安装配置java-jdk

c1331acaeb39594a553b1db01a00606a.png

(2)安装目录如下:

18b79ebf86b6934f35a440c7136fdef8.png

(3)环境变量配置:

7925ac6cb2b42a738b213b7d1486375b.png

d7204951a1f90cab08165613b2ae80ee.png

二、下载kafka

(1)下载kafka2.10-0.9.0.1版本,自带了zookeeper jar包,不用再次下载zookeeper。kafka代理无状态,zookeeper维持集群状态。下载地址:http://mirror.bit.edu.cn/apache/kafka/2.2.0/kafka-2.2.0-src.tgz

(2)安装目录(不要带空格)如下:

3b8d400e763827f3a68b20592b4161f9.png

(3)修改zookeeper和kafka配置文件:

009058df89d9014b595477eb5944beb5.png

cfed89ff0639af077288f0119f01dc63.png

bc78cb1acb3d4b2c15c6df8a27077588.png

(4)按照配置创建这两个目录

(5)cmd启动zookeeper:

cd D:KAFKAkafka_2.10-0.9.0.1kafka_2.10-0.9.0.1kafka_2.10-0.9.0.1binwindowszookeeper-server-start.bat configzookeeper.properties

157a88192daa2c027d3912c1dedda2de.png

(6)再开cmd启动kafka:

cd D:KAFKAkafka_2.10-0.9.0.1kafka_2.10-0.9.0.1kafka_2.10-0.9.0.1binwindowskafka-server-start.bat configserver.properties

64fd79356b9b06faf01520b2d6472e55.png

(7)再开cmd创建topic发送消息:

cd D:KAFKAkafka_2.10-0.9.0.1kafka_2.10-0.9.0.1kafka_2.10-0.9.0.1

#创建topic

binwindowskafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kjTest#列出topic

binwindowskafka-topics.bat --list --zookeeper localhost:2181

#创建生产者

binwindowskafka-console-producer.bat --broker-list localhost:9092 --topic kjTest#发送消息

this isa test

hello

ed720605caf3167f8f99feaa84414cf8.png

(8)再开cmd接收消息:

cd D:KAFKAkafka_2.10-0.9.0.1kafka_2.10-0.9.0.1kafka_2.10-0.9.0.1

#创建消费者

binwindowskafka-console-consumer.bat --zookeeper localhost:2181 --topic kjTest --from-beginning#消费

this isa test

hello

124424d1f54b4670d974e7b1557e5031.png

golang实现Kfaka消息发送:

创建main.go:

package main

import (

"fmt"

"github.com/Shopify/sarama"

"time"

)

//消息写入kafka

func main() {

//初始化配置

config := sarama.NewConfig()

config.Producer.RequiredAcks = sarama.WaitForAll

config.Producer.Partitioner = sarama.NewRandomPartitioner

config.Producer.Return.Successes = true

//生产者

client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)

if err != nil {

fmt.Println("producer close,err:", err)

return

}

defer client.Close()

var n int = 0

for n < 20 {

n++

//创建消息

msg := &sarama.ProducerMessage{}

msg.Topic = "kjTest"

msg.Value = sarama.StringEncoder("this is a good test,hello nola!")

//发送消息

pid, offset, err := client.SendMessage(msg)

if err != nil {

fmt.Println("send message failed,", err)

return

}

fmt.Printf("pid:%v offset:%vn,", pid, offset)

time.Sleep(10 * time.Millisecond)

}

}

消费消息效果:

227f2fe2de02382952344bfbfaff86ed.png

参考博友:

版权声明:本文来源CSDN,感谢博主原创文章,遵循 CC 4.0 by-sa 版权协议,转载请附上原文出处链接和本声明。
原文链接:https://blog.csdn.net/weixin_33256071/article/details/113470327
站方申明:本站部分内容来自社区用户分享,若涉及侵权,请联系站方删除。
  • 发表于 2021-04-10 21:21:23
  • 阅读 ( 1330 )
  • 分类:数据库

0 条评论

请先 登录 后评论

官方社群

GO教程

猜你喜欢