go语言mqtt go语言适合做什么

组件分享之后端组件——基于Golang实现的高性能和弹性的流处理器benthos

近期正在探索前端、后端、系统端各类常用组件与工具，对其一些常见的组件进行再次整理一下，形成标准化组件专题，后续该专题将包含各类语言中的一些常用组件。欢迎大家进行持续关注。

本节我们分享的是基于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。

相关问答

Q1: 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

Q2: apisix和controller如何通信

为什么突然又绕到go语言mqtt了apisix，其实是因为调研nginx-ingress就很容易联想到是不是要替换为其他网关ingress实现，比如apisix-ingress，进而想到go语言mqtt的肯定是apisix；

这些概念到底有什么区别？

apisix 和 apisix-ingress-controller: apisix是一套体系，apisix-ingress-controller是k8s环境下基于go语言开发的一套自定义CRD，他负责响应apisix自定义资源的创建，监听k8s服务资源的创建变更，同步到apisix网关，apisix网关本身由openresty、nginx组成，主要是由Lua语言在nginx执行的各个阶段插入钩子，实现动态化，且基于etcd+灵活的插件机制，让变更更灵活且立即生效；

nginx-ingress和apisix-ingress-controller： nginx-ingress是k8s体系中目前使用最广的路由网关，比如阿里默认的控制台ingress创建就是使用的它；他集合了nginx+nginx-ingress-controller，拉代码可以看出，nginx-ingress-controller完成了与k8sapiserver交互，获取最新的服务配置变更，生成最新nginx配置，除此之外的通讯机制是通过Httppost请求将变更通知到到luanginx侧，ngx.timer在balancer中通过定时器进行动态化

apisix-ingress-controller监控k8s服务变更、自定义CRD及资源创建、将配置同步到apisix；网络请求通过apisix网关作为入口，通过apisix-ingress-controller同步k8s的上游变更

apisix比其他的有啥好？插件灵活，功能多，周边

Q3: web 物联网用什么开发

物联网中最常用的编程语言，即Java，C，C ++，Python，JavaScript和Go。

Javago语言mqtt：物联网技术最流行的编程语言

Java有多个应用领域，从后端编程到Android的移动应用。根据 Eclipse基金会执行的2017年物联网开发者调查，Java首次提供了用于物联网开发的编程语言列表，专门用于网关和云。

使用Java进行物联网开发的一个主要好处是便携性。Java没有任何硬件限制，这意味着您可以在计算机上编写和调试Java代码，并将其部署到几乎任何运行Java虚拟机的设备上。出于这个原因，许多公司选择聘请Java开发人员进行物联网项目。

Cgo语言mqtt：嵌入式设备的关键编程语言

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网络变得更加容易。

go语言mqtt的介绍就聊到这里吧，感谢你花时间阅读本站内容，更多关于go语言适合做什么、go语言mqtt的信息别忘了在本站进行查找喔。