Golang kafka简述和操作（sarama同步异步和消费组）

一、Kafka简述

创新互联建站为客户提供专业的成都网站设计、做网站、程序、域名、空间一条龙服务，提供基于WEB的系统开发. 服务项目涵盖了网页设计、网站程序开发、WEB系统开发、微信二次开发、移动网站建设等网站方面业务。

1. 为什么需要用到消息队列

异步：对比以前的串行同步方式来说，可以在同一时间做更多的事情，提高效率；

解耦：在耦合太高的场景，多个任务要对同一个数据进行操作消费的时候，会导致一个任务的处理因为另一个任务对数据的操作变得及其复杂。

缓冲：当遇到突发大流量的时候，消息队列可以先把所有消息有序保存起来，避免直接作用于系统主体，系统主题始终以一个平稳的速率去消费这些消息。

2.为什么选择kafka呢？

这没有绝对的好坏，看个人需求来选择，我这里就抄了一段他人总结的的优缺点，可见原文

kafka的优点：

1.支持多个生产者和消费者2.支持broker的横向拓展3.副本集机制，实现数据冗余，保证数据不丢失4.通过topic将数据进行分类5.通过分批发送压缩数据的方式，减少数据传输开销，提高吞高量6.支持多种模式的消息7.基于磁盘实现数据的持久化8.高性能的处理信息，在大数据的情况下，可以保证亚秒级的消息延迟9.一个消费者可以支持多种topic的消息10.对CPU和内存的消耗比较小11.对网络开销也比较小12.支持跨数据中心的数据复制13.支持镜像集群

kafka的缺点：

1.由于是批量发送，所以数据达不到真正的实时2.对于mqtt协议不支持3.不支持物联网传感数据直接接入4.只能支持统一分区内消息有序，无法实现全局消息有序5.监控不完善，需要安装插件6.需要配合zookeeper进行元数据管理7.会丢失数据，并且不支持事务8.可能会重复消费数据，消息会乱序，可用保证一个固定的partition内部的消息是有序的，但是一个topic有多个partition的话，就不能保证有序了，需要zookeeper的支持，topic一般需要人工创建，部署和维护一般都比mq高

3. Golang 操作kafka

3.1. kafka的环境

网上有很多搭建kafka环境教程，这里就不再搭建，就展示一下kafka的环境，在kubernetes上进行的搭建，有需要的私我，可以发yaml文件

3.2. 第三方库

github点抗 /Shopify/sarama // kafka主要的库*github点抗 /bsm/sarama-cluster // kafka消费组

3.3. 消费者

单个消费者

funcconsumer(){varwg sync.WaitGroup  consumer, err := sarama.NewConsumer([]string{"172.20.3.13:30901"},nil)iferr !=nil{      fmt.Println("Failed to start consumer: %s", err)return}  partitionList, err := consumer.Partitions("test0")//获得该topic所有的分区iferr !=nil{      fmt.Println("Failed to get the list of partition:, ", err)return}forpartition :=rangepartitionList {      pc, err := consumer.ConsumePartition("test0",int32(partition), sarama.OffsetNewest)iferr !=nil{        fmt.Println("Failed to start consumer for partition %d: %s\n", partition, err)return}      wg.Add(1)gofunc(sarama.PartitionConsumer){//为每个分区开一个go协程去取值formsg :=rangepc.Messages() {//阻塞直到有值发送过来，然后再继续等待fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset,string(msg.Key),string(msg.Value))        }deferpc.AsyncClose()        wg.Done()      }(pc)  }  wg.Wait()}funcmain(){  consumer()}

消费组

funcconsumerCluster(){  groupID :="group-1"config := cluster.NewConfig()  config.Group.Return.Notifications =trueconfig.Consumer.Offsets.CommitInterval =1* time.Second  config.Consumer.Offsets.Initial = sarama.OffsetNewest//初始从最新的offset开始c, err := cluster.NewConsumer(strings.Split("172.20.3.13:30901",","),groupID, strings.Split("test0",","), config)iferr !=nil{      glog.Errorf("Failed open consumer: %v", err)return}deferc.Close()gofunc(c *cluster.Consumer){      errors := c.Errors()      noti := c.Notifications()for{select{caseerr := -errors:            glog.Errorln(err)case-noti:        }      }  }(c)formsg :=rangec.Messages() {      fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset,string(msg.Key),string(msg.Value))      c.MarkOffset(msg,"")//MarkOffset 并不是实时写入kafka，有可能在程序crash时丢掉未提交的offset}}funcmain(){goconsumerCluster()}

3.4. 生产者

同步生产者

packagemainimport("fmt""github点抗 /Shopify/sarama")funcmain(){  config := sarama.NewConfig()  config.Producer.RequiredAcks = sarama.WaitForAll//赋值为-1：这意味着producer在follower副本确认接收到数据后才算一次发送完成。config.Producer.Partitioner = sarama.NewRandomPartitioner//写到随机分区中，默认设置8个分区config.Producer.Return.Successes =truemsg := sarama.ProducerMessage{}  msg.Topic =`test0`msg.Value = sarama.StringEncoder("Hello World!")  client, err := sarama.NewSyncProducer([]string{"172.20.3.13:30901"}, config)iferr !=nil{      fmt.Println("producer close err, ", err)return}deferclient.Close()  pid, offset, err := client.SendMessage(msg)iferr !=nil{      fmt.Println("send message failed, ", err)return}  fmt.Printf("分区ID:%v, offset:%v \n", pid, offset)}

异步生产者

funcasyncProducer(){  config := sarama.NewConfig()  config.Producer.Return.Successes =true//必须有这个选项config.Producer.Timeout =5* time.Second  p, err := sarama.NewAsyncProducer(strings.Split("172.20.3.13:30901",","), config)deferp.Close()iferr !=nil{return}//这个部分一定要写，不然通道会被堵塞gofunc(p sarama.AsyncProducer){      errors := p.Errors()      success := p.Successes()for{select{caseerr := -errors:iferr !=nil{              glog.Errorln(err)            }case-success:        }      }  }(p)for{      v :="async: "+ strconv.Itoa(rand.New(rand.NewSource(time.Now().UnixNano())).Intn(10000))      fmt.Fprintln(os.Stdout, v)      msg := sarama.ProducerMessage{        Topic: topics,        Value: sarama.ByteEncoder(v),      }      p.Input() - msg      time.Sleep(time.Second *1)  }}funcmain(){goasyncProducer()select{      }}

3.5. 结果展示-

同步生产打印：

分区ID:0,offset:90

消费打印：

Partition:0,Offset:90,key:,value:Hello World!

异步生产打印：

async:7272async:7616async:998

消费打印：

Partition:0,Offset:91,key:,value:async:7272Partition:0,Offset:92,key:,value:async:7616Partition:0,Offset:93,key:,value:async:998

为什么golang适合中间件

Golang是一种静态类型的编程语言，具有高效性、安全性和可扩展性。它特别适合用于构建中间件，因为它可以更快地生成和处理数据，而且它可以构建可靠的、可维护的系统。 Golang还具有跨平台的能力，可以在各种操作系统中使用，而且可以使用内置的库和框架来编写大量的代码，从而大大简化了中间件的开发过程。此外，Golang的性能也更好，能够更快地处理大量的数据。因此，Golang对于构建中间件来说是一个非常理想的选择。

协程与异步IO

协程，又称微线程，纤程。英文名 Coroutine 。Python对协程的支持是通过 generator 实现的。在generator中，我们不但可以通过for循环来迭代，还可以不断调用 next()函数 获取由 yield 语句返回的下一个值。但是Python的yield不但可以返回一个值，它还可以接收调用者发出的参数。yield其实是终端当前的函数，返回给调用方。python3中使用yield来实现range，节省内存，提高性能，懒加载的模式。

asyncio是Python 3.4 版本引入的 标准库 ，直接内置了对异步IO的支持。

从Python 3.5 开始引入了新的语法 async 和 await ，用来简化yield的语法：

import asyncio

import threading

async def compute(x, y):

print("Compute %s + %s ..." % (x, y))

print(threading.current_thread().name)

await asyncio.sleep(x + y)

return x + y

async def print_sum(x, y):

result = await compute(x, y)

print("%s + %s = %s" % (x, y, result))

print(threading.current_thread().name)

if __name__ == "__main__":

loop = asyncio.get_event_loop()

tasks = [print_sum(1, 2), print_sum(3, 4)]

loop.run_until_complete(asyncio.wait(tasks))

loop.close()

线程是内核进行抢占式的调度的，这样就确保了每个线程都有执行的机会。而 coroutine 运行在同一个线程中，由语言的运行时中的 EventLoop（事件循环） 来进行调度。和大多数语言一样，在 Python 中，协程的调度是非抢占式的，也就是说一个协程必须主动让出执行机会，其他协程才有机会运行。

让出执行的关键字就是 await。也就是说一个协程如果阻塞了，持续不让出 CPU，那么整个线程就卡住了，没有任何并发。

PS: 作为服务端，event loop最核心的就是IO多路复用技术，所有来自客户端的请求都由IO多路复用函数来处理;作为客户端，event loop的核心在于利用Future对象延迟执行，并使用send函数激发协程,挂起,等待服务端处理完成返回后再调用CallBack函数继续下面的流程

Go语言的协程是 语言本身特性 ，erlang和golang都是采用了CSP(Communicating Sequential Processes)模式(Python中的协程是eventloop模型)，但是erlang是基于进程的消息通信，go是基于goroutine和channel的通信。

Python和Go都引入了消息调度系统模型，来避免锁的影响和进程/线程开销大的问题。

协程从本质上来说是一种用户态的线程，不需要系统来执行抢占式调度，而是在语言层面实现线程的调度 。因为协程 不再使用共享内存/数据 ，而是使用 通信 来共享内存/锁，因为在一个超级大系统里具有无数的锁，共享变量等等会使得整个系统变得无比的臃肿，而通过消息机制来交流，可以使得每个并发的单元都成为一个独立的个体，拥有自己的变量，单元之间变量并不共享，对于单元的输入输出只有消息。开发者只需要关心在一个并发单元的输入与输出的影响，而不需要再考虑类似于修改共享内存/数据对其它程序的影响。

分享名称：go语言支持异步 go异步调用
标题URL：http://scyanting.com/article/ddggipp.html