- xml - AJAX/Jquery XML 解析
- 具有多重继承的 XML 模式
- .net - 枚举序列化 Json 与 XML
- XML 简单类型、简单内容、复杂类型、复杂内容
我正在尝试在 golang 中创建一个 google pubsub 订阅者,我一次接收 100 条消息,然后将它们写入 influx。我正在尝试使用 channel 来执行此操作:
package main
import (
"os"
"fmt"
"cloud.google.com/go/pubsub"
"log"
"sync"
"golang.org/x/net/context"
"encoding/json"
clnt "github.com/influxdata/influxdb/client/v2"
"time"
)
type SensorData struct {
Pressure float64 `json:"pressure"`
Temperature float64 `json:"temperature"`
Dewpoint float64 `json:"dewpoint"`
Timecollected int64 `json:"timecollected"`
Latitude float64 `json:"latitude"`
Longitude float64 `json:"longitude"`
Humidity float64 `json:"humidity"`
SensorID string `json:"sensorId"`
Zipcode int `json:"zipcode"`
Warehouse string `json:"warehouse"`
Area string `json:"area"`
}
type SensorPoints struct {
SensorData []SensorData
}
func main () {
messages := make(chan SensorData, 100)
// Create a new Influx HTTPClient
c, err := clnt.NewHTTPClient(clnt.HTTPConfig{
Addr: "http://localhost:8086",
Username: "user",
Password: "pass",
})
if err != nil {
log.Fatal(err)
}
// Create pubsub subscriber
ctx := context.Background()
proj := os.Getenv("GOOGLE_CLOUD_PROJECT")
if proj == "" {
fmt.Fprintf(os.Stderr, "GOOGLE_CLOUD_PROJECT environment variable must be set.\n")
os.Exit(1)
}
client, err := pubsub.NewClient(ctx, proj)
if err != nil {
log.Fatalf("Could not create pubsub Client: %v", err)
}
const sub = "influxwriter"
//create influx a blank batchpoint set
bp, err := clnt.NewBatchPoints(clnt.BatchPointsConfig{
Database: "sensordata",
Precision: "s",
})
if err != nil {
log.Fatal(err)
}
// Pull messages via the subscription.
go pullMsgs(client, sub, messages)
if err != nil {
log.Fatal(err)
}
writeInflux(messages, bp)
c.Close()
}
func pullMsgs(client *pubsub.Client, name string, messages chan<- SensorData) {
ctx := context.Background()
// [START pubsub_subscriber_async_pull]
// [START pubsub_quickstart_subscriber]
// Consume 10 messages.
var mu sync.Mutex
var sensorinfos SensorPoints
sensorinfo := &SensorData{}
received := 0
sub := client.Subscription(name)
cctx, _ := context.WithCancel(ctx)
err := sub.Receive(cctx, func(ctx context.Context, msg *pubsub.Message) {
msg.Ack()
json.Unmarshal(msg.Data, sensorinfo)
//fmt.Println(string(msg.Data))
//fmt.Println(sensorinfo.SensorID)
sensorinfos.SensorData = append(sensorinfos.SensorData, *sensorinfo)
mu.Lock()
defer mu.Unlock()
received++
fmt.Println("rcv: ", received)
messages <- *sensorinfo
})
if err != nil {
fmt.Println(err)
}
// [END pubsub_subscriber_async_pull]
// [END pubsub_quickstart_subscriber]
}
func writeInflux(sensorpoints <- chan SensorData, bp clnt.BatchPoints) {
for p := range sensorpoints {
// Create a point and add to batch
tags := map[string]string{
"sensorId": p.SensorID,
"warehouse": p.Warehouse,
"area": p.Area,
"zipcode": string(p.Zipcode),
}
fields := map[string]interface{}{
"pressure": p.Pressure,
"humidity": p.Humidity,
"temperature": p.Temperature,
"dewpoint": p.Dewpoint,
"longitude": p.Longitude,
"latitude": p.Latitude,
}
pt, err := clnt.NewPoint("sensordata", tags, fields, time.Unix(p.Timecollected, 0))
if err != nil {
log.Fatal(err)
}
bp.AddPoint(pt)
}
}
但它并没有看到每个人都通过初始的 pullMsgs
函数,只是继续打印那里的输出:
rcv: 1
rcv: 2
rcv: 3
rcv: 4
rcv: 5
rcv: 6
rcv: 7
我认为一旦 channel 满了,它应该阻塞直到 channel 被清空
这是我用作引用的 pubsub 拉取代码。
最佳答案
当您在 channel 上发送了所需数量的消息后,关闭 channel 并取消上下文。尝试使用 the documentation 中演示的技术在一些消息后取消。由于您的缓冲区是 100,并且您尝试一次使用 100 条消息,所以这就是数字。如果您希望程序退出,请关闭 channel ,以便 writeInflux
中的 for e := range ch
循环到达停止点并且不会阻塞等待更多元素将添加到 channel 。
在 the Go pubsub API doc 中注意这一点:
To terminate a call to Receive, cancel its context.
这不是让您的主 goroutine 停滞的原因,而是您的 pullMsgs
goroutine 在没有取消的情况下不会自行退出。
此外,检查 Unmarshal
上的错误。如果您不想在代码的此时处理解码错误,请考虑更改 channel 类型并改为发送 msg
或 msg.Data
并在 channel 接收时解码。
cctx, cancel := context.WithCancel(ctx)
err := sub.Receive(cctx, func(ctx context.Context, msg *pubsub.Message) {
msg.Ack()
err := json.Unmarshal(msg.Data, sensorinfo)
if err != nil {
fmt.Printf("Failed to unmarshal: %s\n", err)
}
mu.Lock()
defer mu.Unlock()
received++
fmt.Println("rcv: ", received)
messages <- *sensorinfo
if received == 100 {
close(messages) // no more messages will be sent on channel
cancel()
}
关于go - 将 channel 与 google pubsub 民意调查订阅者一起使用,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51717626/
我想要的是能够在输入获得焦点或失去焦点时执行某些操作(两个事件)。 我尝试了以下方法,但这按事件单独工作(单独编码时):仅在焦点上,或仅在失去焦点时。 另外,我希望它尽可能跨平台(包括触摸设备),这是
我分别研究了TableView的Filtering和Pagination。 过滤: this帖子帮助我满足了我的需要 分页: this , this帖子也帮助了我 我想像这样将它们组合在一起: 详情-
我是 TDD 方法的新手,所以我想知道是否有人经历过这种机智可以启发我一点。我想获得一些关于如何一起使用 UML 和 TDD 方法的线索。 我已经习惯了:用 UML 设计 --> 生成骨架类(然后保持
我尝试使用入口点和 cmd 设置 Docker。 FROM debian:stretch RUN apt-get update && \ apt install gnupg ca-certificat
我想要一个 Class 对象,但我想强制它所代表的任何类扩展类 A 并实现接口(interface) B。 我能做到: Class 或者: Class 但我不能两者兼得。有办法做到这一点吗? 最佳答案
我是 Rubymine 的长期用户。 Rubymine 非常适合基于 html 的 Rails 应用程序,但我现在正在做更多的 SPA 客户端工作(例如 javascript/react)。我发现我真
我注意到我使用的某个脚本依赖于原型(prototype)。 (Lightbox 2) 它会与 jQuery 在同一页面上一起工作吗?有没有办法确保它们不冲突? 最佳答案 可以,但你需要采取 speci
我需要对表中显示的数据进行分页并通过 ajax 调用获取它 - 这是我通过使用具有以下配置的 dataTables 插件来完成的 - bServerSide : true; sAjaxSource :
我是 gtk 新手,所以想知道在 C 语言中归档和 gtk 是否可以一起使用?例如,我可以从 .txt 文件中读取,然后在相同的代码中使用 gtk 在标签或其他内容中显示它吗?如果是,怎么办? 谢谢!
有没有人设法得到Bck2Brwsr最近与 Java 8/JavaFX 8 一起工作?有没有兼容的机会?我找不到太多关于它的信息,也没有一个好的起点。使用给定的 Maven archetype我遇到了几
在我的应用程序中,用户通过 openid(与 stackoverflow 相同)登录/注销。 我想通过 oauth 向第三方应用程序开放我的应用程序。 如何创建我的 openid-consumer 应
我在启动和运行 Hibernate 和 Spring 时遇到一些问题。我有一个网络服务器项目,它使用了其他几个具有持久实体的项目。我遇到的问题是,对于存储在 WEB-INF/libs 内的另一个 ja
我有 @ControllerAdvice 类,它处理一组异常。我们还有一些其他异常,这些异常用 @ResponseStatus 注释进行注释。为了结合这两种方法,我们使用博客文章中描述的技术:http
我想在屏幕上使用进度条而不是 progressDialog。 我在我的 XML View 文件中插入了一个进度条,我想让它在加载时显示并在不加载时禁用它。 所以我使用的是可见的,但它发生了,所以其余的
CREATE TABLE `users` ( `id` int(11) AUTO_INCREMENT, `academicdegree` varchar(255),
IN() 中使用的查询返回:1, 2。然而,整个查询返回 0 行,这是不可能的,因为它们存在。我在这里做错了什么? SELECT DISTINCT li.auto_id FROM links
亲们, 我如何在使用 Jade 生成的表单上实现 jQuery 样式?我想做的是美化 表单并使它们可点击。我在 UI 方面很糟糕。期间。 我如何在表单上实现这个可选择的方法? http://jquer
按照目前的情况,这个问题不适合我们的问答形式。我们希望答案得到事实、引用或专业知识的支持,但这个问题可能会引发辩论、争论、投票或扩展讨论。如果您觉得这个问题可以改进并可能重新打开,visit the
我可以: auto o1 = new Content; 但不能: std::shared_ptr o1(new Content); std::unique_ptr o1(new Content); 我
关闭。这个问题需要更多focused .它目前不接受答案。 想改进这个问题吗? 更新问题,使其只关注一个问题 editing this post . 关闭 4 年前。 Improve this qu
我是一名优秀的程序员,十分优秀!