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

一、Kafka简述

按需制作可以根据自己的需求进行定制，成都网站设计、做网站构思过程中功能建设理应排到主要部位公司成都网站设计、做网站的运用实际效果公司网站制作网站建立与制做的实际意义

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.com/Shopify/sarama // kafka主要的库*github.com/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.com/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

EMQ功能使用(一) 实现MQTTS协议

前言

EMQ是带有SSL功能的，需要进行简单的配置，才能使用。下面就简单说一下如何实现自签证书。

利用OpenSSL签发证书

配置到EMQX的emqx.conf

如果启用单向认证的话，客户端不需要证书都可以连接。这里的listener.ssl.external.fail_if_no_peer_cert = true 注释掉就启用单向认证。启用双向认证。那么客户端就必须导入CA和client的证书才可以连接。

重新启动EMQX

工具测试

我这里使用EMQ官方出品的MQTTX工具

参考：

【内部分享】MQTT协议解读及使用经验

时间：2018-07-26

Q: 什么是网络连接？

A: 网络连接是传输层定义的概念，在传输层以下只存在网络数据包的相互交换。

所谓连接，其实也不是在网络上有一条真实存在的数据通道。只要通信双方在一段时间内持续保持数据包交换，就可以视为双方建立的连接并没有断开。

连接的建立是依托于TCP协议的三次握手，一旦连接已经建立完毕，通信双方就可以复用这条虚拟通道进行数据交换。如果连接保持长时间工作一直没有被中断，那么这样的TCP连接就俗称为长连接。

Message Queue Telemetry Transport ，中文直译： 消息队列遥测传输协议 。

在MQTT协议被设计出来的年代，还没有物联网这么时髦的词汇，当年叫做 遥测设备 。

MQTT协议真正开始声名鹊起的原因，是其正好恰恰踩中移动互联网发展的节拍，为消息推送场景提供了一个既简便又具有良好扩展性的现成解决方案。

可以看出，MQTT对消息头的规定十分精简， 固定头部占用空间大小仅为1个字节 ，一个最小的报文占用的空间也 只有两个字节 （带一字节的长度标识位）。

这也是MQTT协议针对不稳定及带宽低下的网络环境做出的特定设计 - - - - 尽可能地节省一切不必要的网络开销 。

Q：为什么MQTT协议需要心跳报文(PINGREQ, PINGRESP)来维护连接状态，只监控该TCP的连接状态是否可以实现目的？

A: TCP数据传输默认的超时时间过长，不符合应用层上细粒度的要求。

TCP数据传输超时的情况可分成三种： 服务端断开 、 客户端断开 、 中间网络断开 。

在前两种场景下，若断开操作是一方主动发起的，即表示为TCP连接正常结束，双方走四次挥手流程；若程序异常结束，则会触发被动断开事件，通信另一方也能立刻感知到本次连接所打开的 Socket 出现中断异常。

唯独中间网络的状态是通信双方不能掌握的。 在Linux系统下 ，TCP的连接超时由内核参数来控制，如果通信中的一方没有得到及时回复，默认会主动再尝试 6次 。如果还没有得到及时回应，那么其才会认定本次数据超时。

连带首次发包与六次重试，Linux系统下这7次发包的超时时间分别为 2的0次方 至 2的6次方 ，即1秒、2秒、4秒、8秒、16秒、32秒、64秒，一共127秒。MQTT协议认为如此长的超时时间对应用层而言粒度太大，因此其在应用层上还单独设计属于自身的心跳响应控制。常见的MQTT连接超时多被设定为 60秒 。

扩展知识 - TCP的KeepAlive机制：

由通信中的 报文标识符 ( Packet Identifier )传达。

Q：仅Publish与Pubrec能保证消息只被投递一次吗？

A： 业务上可以实现，但MQTT协议并没有如此设计，原因如下：

每个消息都会拥有属于自己的报文标识符，但如果需要两次数据交换就实现消息仅只收到一次，就需要通信双方记录下每次使用的报文标识符，并且在处理每一条消息时都需要去重处理，以防消息被重复消费。

但MQTT协议最初被设计的工作对象是轻量级物联设备，为此在协议的设计中报文标识符被约定为 可重用 ，以减少对设备性能的消耗，换回的代价不得不使用四次网络数据交换，才能确保消息正好被消费一次。

Q：两个不同客户端在发布与订阅同一Topic下的消息时，都可以提出通信Qos要求，此时以哪项为基准？

A： 伪命题，故意在分享时埋下坑，等人来踩。

两个不同客户端的通信是需要 Broker 进行中转，而不是直连。因此，通信中存在两个不同的会话，双方的Qos要求仅仅作用于它们与 Broker 之间的会话，最终的Qos基准只会向最低要求方看齐。

例：遗嘱消息的正确使用方式可参考此篇文章：

虽然可以借助 Retain Message 实现绑定一条消息至某个Topic，以达到消息的暂时保留目的。

但首先 Retain Message 并不是为存储场景而设计的，再次MQTT协议并没有对消息的持久化作出规定，也就是说Broker重启后，现有保留消息也将丢失。

Q：两种特殊消息的使用场景？

A： 遗嘱消息，多用于客户端间获取互相之间异常断线的消息通知；

保留消息，可保存 最近一条 广播通知，多用于公告栏信息的发布。

Eclipse Mosquitto ：MQTT协议的最小集实现

有 EMQ , HiveMQ , RabbitMQ MQTT Adapter 等。

Qos=2 消息保障的网络I/O次数过多，如果不是必需，尽少在程序里使用此类消息。

毕竟当初其设计的目的是为了减少设备的性能占用，但若应用场景并不是物联网，而是用于手机、电脑或浏览器端等现在已不缺性能的设备上，最好在报文体中，使用UUID生成全局唯一的消息ID，然后自行在业务解析中判断此报文是否被消费过。

或者，业务方在处理消息时保证其被消费的幂等性，也可消除重复消息对系统带来的影响。

正如MQTT协议并没有依赖TCP连接状态，自己在应用层协议上实现心跳报文来控制连接状态，业务方作为MQTT协议的使用者，也不要完全依赖协议的工作状态，而是依托MQTT协议建立属于业务本身的信息汇报机制，以加强系统的稳健性。

Retain Message 可视为客户端主动拉取的行为。如果业务系统采用 HTTP+MQTT 双协议描述业务过程，主动拉取的操作也可使用 HTTP 请求替代。

作为一个长连接型的应用，上线前需要根据业务量级，评估对操作系统 端口数 与 文件描述符 的占用要求，以防服务器资源被打满。

在服务端的配置文件和客户端的连接参数中，都拥有 max_inflight_messages 此项配置，来维护 Qos=1 or 2 消息是否被成功消费的状态。

MQTT 最初被设计为物联网级的通信协议，因此此参数的默认配额较小（大多数情况下被限制到10至20）。

但如果将MQTT协议应用至手机、PC或Web端的推送场景时，硬件性能已不在是瓶颈，在实际使用中推荐把此参数调大。

Mosquitto提供Bridge功能，需要我们自己配置。

Bridge 意为桥接，当我们把两台Broker桥接在一起时，只需要修改一台Broker的配置，填上另一台Broker的运行地址。前一台Broker将作为客户端发布与订阅后一台Broker的所有Topic，实现消息互通的目的。

桥接带来的问题有以下几点：

我的建议：

Websockets协议被设计的目的是为浏览器提供一个全双工的通信协议，方便实现消息推送功能。

在Websockets协议被设计出来前，受限于HTTP协议的一问一答模型，消息的推送只能靠轮询来实现，在资源消耗与时效性保障上，均难以达到令人满意的效果。

Websockets协议复用了HTTP协议的头部信息，告知浏览器接下来的操作将触发协议升级，然后通信双方继续复用HTTP的Header，但报文内容已转变为双方均接受的新协议的格式。

Websockets协议改进了网页浏览中的消息推送的方式，因此被广泛应用在聊天、支付通知等实时性要求比较高的场合下。

MQTT协议重点在于 消息队列的实现，其对消息投递的方式作出约定，并提供一些额外的通信保障 。

MQTT可采取原生的TCP实现，也有基于Websockets的实现版本。当然后者在网络字节的利用率上，不如前者那么精简。但浏览器端无法直接使用TCP协议，所以就只能基于Websockets协议开发。

不过基于Websockets的应用也有方便之处：一是证书不需要额外配置，直接与网站共用一套基础设施；二是可使用 Nginx 等工具管理流量，与普通HTTP流量可共用一套配置方法。

MQTT非常适合入门，原因如下：

实际的应用场景远比理想中的复杂，无法一招走遍天下，必须做好取舍。

MQTT协议在这方面做得很优秀，以后工作中可以作为参考，设计好自己负责的业务系统。

ios 利用mqttclient库写客户端怎么添加ssl证书

iOS7以下版本的国际化方法 创建一个Localizable.strings文件 code中用字符串的地方使用NSLocalizedString，这与系统语言相关 时间显示，数字，金融与地区相关，所以需要各类NSFormater, 如NSDateFormatter, NSNumberFormatter … 用命令将所有NSLocalizedString返回的字符串格式化到Localizable.strings里。命令行进入工程目录（我的工程名是LocalizationTest），运行下面命令： find ./ -name "*.m" -print0 xargs -0 genstrings -o LocalizationTest/en.lproj 其中LocalizationTest/en.lproj根据自己的目录修改签SSL证书问题，你可要到沃通CA申请一张免费SSL证书来使用。

mqtt使用WebSocket over TLS(wss)握手失败

由于网页运行在https上，所以连接mqtt只能用wss，但是使用自签证书一直显示 1015 TLS_HANDSHAKE ，可以判断为认证阶段不通过。在MQTT.fx上面则提示证书非法。后面找了很多资料，终于在一个回到里面找到答案，就记录下来。

MQTT客户端软件MQTT.fx的使用详解

原地址：

原文链接：

MQTT客户端软件MQTT.fx的使用详解

使用说明

mqtt.fx打开后的主页面如下：

点击齿轮进行连接设置

本地连接设置：

用户信息设置：

SSL安全证书设置：

网络代理设置：

遗嘱设置：

连接测试

1、启动mosquitto

地址，下一步配置使用

2、在主机中打开MQTT.FX软件

设置连接信息

IP为mosquitto所在的IP，端口号默认为1883。

点击进行连接

连接成功以后可以进行发布订阅。