近期正在探索前端、后端、系统端各类常用组件与工具，对其一些常见的组件进行再次整理一下，形成标准化组件专题，后续该专题将包含各类语言中的一些常用组件。欢迎大家进行持续关注。
成都创新互联-专业网站定制、快速模板网站建设、高性价比石嘴山网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式石嘴山网站制作公司更省心,省钱,快速模板网站建设找我们，业务覆盖石嘴山地区。费用合理售后完善，10余年实体公司更值得信赖。
本节我们分享的是基于Golang实现的高性能和弹性的流处理器 benthos ，它能够以各种代理模式连接各种 源 和 接收器，并对有效负载执行 水合、浓缩、转换和过滤 。
它带有 强大的映射语言 ，易于部署和监控，并且可以作为静态二进制文件、docker 映像或 无服务器函数 放入您的管道，使其成为云原生。
Benthos 是完全声明性的，流管道在单个配置文件中定义，允许您指定连接器和处理阶段列表：
Apache Pulsar, AWS (DynamoDB, Kinesis, S3, SQS, SNS), Azure (Blob storage, Queue storage, Table storage), Cassandra, Elasticsearch, File, GCP (Pub/Sub, Cloud storage), HDFS, HTTP (server and client, including websockets), Kafka, Memcached, MQTT, Nanomsg, NATS, NATS JetStream, NATS Streaming, NSQ, AMQP 0.91 (RabbitMQ), AMQP 1, Redis (streams, list, pubsub, hashes), MongoDB, SQL (MySQL, PostgreSQL, Clickhouse, MSSQL), Stdin/Stdout, TCP UDP, sockets and ZMQ4.
1、docker安装
具体使用方式可以参见该 文档
有关如何配置更高级的流处理概念（例如流连接、扩充工作流等）的指导，请查看 说明书部分。
有关在 Go 中构建您自己的自定义插件的指导，请查看 公共 API。
物联网中最常用的编程语言，即Java，C，C ++，Python，JavaScript和Go。
Java：物联网技术最流行的编程语言
Java有多个应用领域，从后端编程到Android的移动应用。根据 Eclipse基金会执行的2017年物联网开发者调查，Java首次提供了用于物联网开发的编程语言列表，专门用于网关和云。
使用Java进行物联网开发的一个主要好处是便携性。Java没有任何硬件限制，这意味着您可以在计算机上编写和调试Java代码，并将其部署到几乎任何运行Java虚拟机的设备上。出于这个原因，许多公司选择聘请Java开发人员进行物联网项目。
C：嵌入式设备的关键编程语言
C编程语言接下来成为物联网IoT堆栈最喜欢的语言。然而，根据Eclipse基金会的说法，它被认为是受限设备开发的领先技术。
该编程语言提供对低级硬件API的直接访问。由于其与机器语言的相似性，C非常快速且灵活，使其成为处理能力有限的物联网系统的完美选择。
C ++：Linux的第一语言
与其前身C一样，C ++已广泛用于嵌入式系统开发。但是，C ++的主要优势在于处理能力，在任务更加复杂时使其成为C的有用替代方案。
C ++最适合编写硬件特定的代码。它可与Linux，第一大物联网技术操作系统配合使用。但是，与Java相比，它具有有限的可移植性。
Python：面向数据的物联网系统的解决方案
作为最受欢迎的网络编程语言之一，以及科学计算的前沿技术，Python在物联网开发中也获得了巨大的推动力。 对于数据密集型应用程序，Python是一个不错的选择，特别是在管理和组织复杂数据时。
JavaScript：事件驱动物联网应用的最佳解决方案
根据年度StackOverflow开发者调查显示，JavaScript是过去五年来最流行的编程语言之一，是现代Web开发中的核心技术。
在许多其他应用领域中，JavaScript是物联网编程语言中最常用的构建事件驱动系统。它可以管理连接设备的大型网络，并且在需要处理多个任务而无需等待其他任务完成时可以胜任。JavaScript对IoT的主要优势之一是非常节约资源。
Go：坚固的技术堆栈为复杂的物联网网络提供动力
Go是一款开源编程语言，由Google创建。尽管它不能像语言那样拥有同样广泛的用途，但我们之前专注于这一点，它是在您的物联网系统内建立通信层的强大技术。
Go语言关于物联网的主要优势是并发性和同时运行多个进程（数据输入和输出）的能力。这使得构建由多个传感器和设备组成的复杂IoT网络变得更加容易。
最近在着手研究使用网页控制硬件的交互过程，发现mqtt协议有很多种语言的封装，整个思路也很清晰，就是一个客户端进行发布和订阅+服务器中间代理的过程，于是开始学习使用MQTT框架搭建交互平台。
客户端使用了MQTT.js，服务器使用mosca（基于nodejs）。
QoS在MQTT中有（摘自 MQ 遥测传输 (MQTT) V3.1 协议规范 ）：
MQTT.js只是支持了MQTT协议，并没有支持QoS，也就是说，只支持最低级别的“至多一次”（QoS0）。
在MQTT协议中，一个MQTT数据包由：固定头（Fixed header）、可变头（Variable header）、消息体（payload）三部分构成。
这里只了解消息体，固定头和可变头并不需要我们手动写。
payload消息体包含CONNECT、SUBSCRIBE、SUBACK、UNSUBSCRIBE四种类型的消息：
（1）Connect 与服务器建立连接。
（2）Disconnect 与服务器断开TCP/IP会话。
（3）Subscribe 订阅。
（4）UnSubscribe 取消订阅。
（5）Publish 发送消息请求，发送完成后返回应用程序线程。
二、实现（具体的API稍后呈现）
MQTT跟WebSocket关系不大。他们不是在一个层级的。
WebSocket 很多网站使用轮询实现推送技术。轮询是在特定的的时间间隔（比如1秒），由浏览器对服务器发出HTTP request，然后由服务器返回最新的数据给浏览器。轮询的缺点很明显，浏览器需要不断的向服务器发出请求，然而HTTP请求的header是非常长的，而实际传输的数据可能很小，这就造成了带宽和服务器资源的浪费。
Comet使用了AJAX改进了轮询，可以实现双向通信。但是Comet依然需要发出请求，而且在Comet中，普遍采用了长链接，这也会大量消耗服务器带宽和资源。
于是，WebSocket协议应运而生。 浏览器通过 JavaScript 向服务器发出建立 WebSocket 连接的请求，连接建立以后，客户端和服务器通过 TCP 连接直接交换数据。WebSocket 连接本质上是一个 TCP 连接。
WebSocket在数据传输的稳定性和数据传输量的大小方面，具有很大的性能优势。Websocket.org 比较了轮询和WebSocket的性能优势：
HTTP 轮训每次需要返回871个字节，websocket每次只需要2个字节
Use Case A： 1,000个客户端每秒接受一个message，网络吞吐量 （2*1,000）=2,000 bytes = 16,000 每秒bits
Use Case B： 10,000个客户端每秒接受一个message，网络吞吐量 （2*10,000）=20,000 bytes = 160,000 每秒bits
Use Case C： 100,000个客户端每秒接受一个message，网络吞吐量 （2*100,000）=200,000 bytes = 1,600,000 每秒bits
MQTT 协议是为大量计算能力有限，且工作在低带宽、不可靠的网络的远程传感器和控制设备通讯而设计的协议，它具有以下主要的几项特性：
非常小的通信开销（最小的消息大小为 2 字节），小型传输，开销很小（固定长度的头部是 2 字节），协议交换最小化，以降低网络流量。
支持各种流行编程语言（包括 C，Java，Ruby，Python 等等）且易于使用的客户端；
使用发布 / 订阅消息模式，提供一对多的消息发布，解除应用程序耦合。
对负载内容屏蔽的消息传输。
使用 TCP/IP 提供网络连接。
有三种消息发布服务质量，让消息能按需到达目的地，适应在不稳定工作的网络传输需求 ：
"至多一次"，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。
"至少一次"，确保消息到达，但消息重复可能会发生。
"只有一次"，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果。
一、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