編輯：在我添加了一小部分我正在使用的文件（7 GB）并嘗試運行該程序后，我可以看到：fatal error: all goroutines are asleep - deadlock!goroutine 1 [chan receive]:main.main() /media/developer/golang/manual/examples/sp/v2/sp.v2.go:71 +0x4a9exit status 2情況：我是 GO 的新手，所以如果我的問題真的很簡單，我很抱歉。我正在嘗試流式傳輸xml文件、拆分文檔，然后在不同的 GO 例程中解析它們。我正在使用的 XML 文件示例：<?xml version="1.0" encoding="UTF-8"?><osm version="0.6" generator="CGImap 0.0.2"> <relation id="56688" user="kmvar" uid="56190" visible="true" version="28" changeset="6947637" timestamp="2011-01-12T14:23:49Z"> <member type="node" ref="294942404" role=""/> <member type="node" ref="364933006" role=""/> <tag k="name" v="Küstenbus Linie 123"/> <tag k="network" v="VVW"/> <tag k="route" v="bus"/> <tag k="type" v="route"/> </relation> <relation id="98367" user="jdifh" uid="92834" visible="true" version="28" changeset="6947637" timestamp="2011-01-12T14:23:49Z"> <member type="node" ref="294942404" role=""/> <member type="way" ref="4579143" role=""/> <member type="node" ref="249673494" role=""/> <tag k="name" v="Küstenbus Linie 123"/> <tag k="network" v="VVW"/> <tag k="operator" v="Regionalverkehr Küste"/> <tag k="ref" v="123"/> </relation> <relation id="72947" user="schsu" uid="92374" visible="true" version="28" changeset="6947637" timestamp="2011-01-12T14:23:49Z"> <member type="node" ref="294942404" role=""/> <tag k="name" v="Küstenbus Linie 123"/> <tag k="type" v="route"/> </relation></osm>它應該打印每個結構，但我什么也沒看到。該程序只是掛起。我測試了 streamer 和 splitter（只是在將消息發送到通道之前添加fmt.Println(rs)到函數parseRelation中）。我可以看到結構。因此，問題在于發送和接收消息。問題：我不知道如何解決這個問題。嘗試更改頻道中消息的類型（從RS到string）并且每次只發送一個字符串。但它也沒有幫助（我什么也看不到）
2 回答
呼喚遠方
TA貢獻1856條經驗 獲得超11個贊
首先，讓我們解決這個問題：您不能逐行解析 XML。您很幸運，您的文件恰好是每行一個標簽，但這不能想當然。您必須解析整個 XML 文檔。
通過逐行處理，您試圖將<tag>and<member>推入為<relation>.?相反，使用xml.NewDecoder并讓它為您處理文件。
package main
import (
? ? "encoding/xml"
? ? "fmt"
? ? "os"
? ? "log"
)
type Osm struct {
? ? XMLName? ? ?xml.Name? ? `xml:"osm"`
? ? Relations? ?[]Relation? `xml:"relation"`
}
type Relation struct {
? ? XMLName? ? ?xml.Name? ? `xml:"relation"`
? ? ID? ? ? ? ? string? ? ? `xml:"id,attr"`
? ? User? ? ? ? string? ? ? `xml:"user,attr"`
? ? Uid? ? ? ? ?string? ? ? `xml:"uid,attr"`
? ? Members? ? ?[]Member? ? `xml:"member"`
? ? Tags? ? ? ? []Tag? ? ? ?`xml:"tag"`
}
type Member struct {
? ? XMLName? ? ?xml.Name? ? `xml:"member"`
? ? Ref? ? ? ? ?string? ? ? `xml:"ref,attr"`
? ? Type? ? ? ? string? ? ? `xml:"type,attr"`
? ? Role? ? ? ? string? ? ? `xml:"role,attr"`
}
type Tag struct {
? ? XMLName? ? ?xml.Name? ? `xml:"tag"`
? ? Key? ? ? ? ?string? ? ? `xml:"k,attr"`
? ? Value? ? ? ?string? ? ? `xml:"v,attr"`
}
func main() {
? ? reader, err := os.Open("test.xml")
? ? if err != nil {
? ? ? ? log.Fatal(err)
? ? }
? ? defer reader.Close()
? ? decoder := xml.NewDecoder(reader)
? ? osm := &Osm{}
? ? err = decoder.Decode(&osm)
? ? if err != nil {
? ? ? ? log.Fatal(err)
? ? }
? ? fmt.Println(osm)
}
Osm其他結構類似于您期望的 XML 模式。decoder.Decode(&osm)應用該架構。
答案的其余部分將僅涵蓋通道和 goroutines 的使用。XML 部分將被刪除。
如果你添加一些調試語句，你會發現它parseRelation從未被調用，這意味著它channel是空的并且fmt.Println(<- channel)等待一個永遠不會關閉的空通道。所以一旦你完成處理，關閉通道。
? for {
? ? p2Rc, err := reader.ReadSlice('\n')
? ? ...
? }
? close(channel)
現在我們得到{ [] []}5973974 次。
for i := 0; i < 5973974; i++ {
? fmt.Println(<- channel)
}
這是嘗試從通道讀取 5973974 次。這違背了渠道的意義。相反，使用range.
for thing := range channel {
? ? fmt.Println(thing)
}
現在至少它完成了！
但是有一個新問題。如果它真的找到了一個東西，比如如果你改成if p2Rc[2] == 114 {，if p2Rc[2] == 32 {你會得到一個panic: send on closed channel. 這是因為parseRelation與閱讀器并行運行，并且可能會在主要閱讀代碼完成并關閉通道后嘗試寫入。在關閉頻道之前，您必須確保使用該頻道的每個人都已完成。
要解決此問題，需要進行相當大的重新設計。
下面是一個簡單程序的示例，該程序從文件中讀取行，將它們放入通道，并讓工作人員從該通道讀取。
func main() {
? ? reader, err := os.Open("test.xml")
? ? if err != nil {
? ? ? ? log.Fatal(err)
? ? }
? ? defer reader.Close()
? ? // Use the simpler bufio.Scanner
? ? scanner := bufio.NewScanner(reader)
? ? // A buffered channel for input
? ? lines := make(chan string, 10)
? ? // Work on the lines
? ? go func() {
? ? ? ? for line := range lines {
? ? ? ? ? ? fmt.Println(line)
? ? ? ? }
? ? }()
? ? // Read lines into the channel
? ? for scanner.Scan() {
? ? ? ? lines <- scanner.Text()
? ? }
? ? if err := scanner.Err(); err != nil {
? ? ? ? log.Fatal(err)
? ? }
? ? // When main exits, channels gracefully close.
}
這很好用，因為main它很特殊并且在退出時會清理通道。但是如果 reader 和 writer 都是 goroutines 呢？
// A buffered channel for input
lines := make(chan string, 10)
// Work on the lines
go func() {
? ? for line := range lines {
? ? ? ? fmt.Println(line)
? ? }
}()
// Read lines into the channel
go func() {
? ? for scanner.Scan() {
? ? ? ? lines <- scanner.Text()
? ? }
? ? if err := scanner.Err(); err != nil {
? ? ? ? log.Fatal(err)
? ? }
}()
空的。main在 goroutines 可以完成他們的工作之前退出并關閉通道。我們需要一種方法讓我們知道main要等到處理完成。有幾種方法可以做到這一點。一個是與另一個通道同步處理。
// A buffered channel for input
lines := make(chan string, 10)
// A channel for main to wait for
done := make(chan bool, 1)
// Work on the lines
go func() {
? ? for line := range lines {
? ? ? ? fmt.Println(line)
? ? }
? ? // Indicate the worker is done
? ? done <- true
}()
// Read lines into the channel
go func() {
? ? // Explicitly close `lines` when we're done so the workers can finish
? ? defer close(lines)
? ? for scanner.Scan() {
? ? ? ? lines <- scanner.Text()
? ? }
? ? if err := scanner.Err(); err != nil {
? ? ? ? log.Fatal(err)
? ? }
}()
// main will wait until there's something to read from `done`
<-done
現在main將觸發 reader 和 worker goroutines 并緩沖等待done. 閱讀器將填充lines直到完成閱讀，然后將其關閉。同時，工作人員將在完成讀取后讀取lines和寫入。done
另一種選擇是使用sync.WaitGroup。
// A buffered channel for input
lines := make(chan string, 10)
var wg sync.WaitGroup
// Note that there is one more thing to wait for
wg.Add(1)
go func() {
? ? // Tell the WaitGroup we're done
? ? defer wg.Done()
? ? for line := range lines {
? ? ? ? fmt.Println(line)
? ? }
}()
// Read lines into the channel
go func() {
? ? defer close(lines)
? ? for scanner.Scan() {
? ? ? ? lines <- scanner.Text()
? ? }
? ? if err := scanner.Err(); err != nil {
? ? ? ? log.Fatal(err)
? ? }
}()
// Wait until everything in the WaitGroup is done
wg.Wait()
和以前一樣，main啟動 reader 和 worker goroutine，但現在它在啟動 worker 之前將 1 添加到 WaitGroup。然后等待wg.Wait()返回。閱讀器與以前一樣工作，lines完成后關閉通道。工作人員現在wg.Done()在完成遞減 WaitGroup 的計數并允許wg.Wait()返回時調用。
每種技術都有優點和缺點。done更靈活，鏈條更好，如果你能把頭繞在它身上會更安全。WaitGroups 更簡單，更容易理解，但要求每個 goroutine 共享一個變量。
如果我們想添加到這個處理鏈中，我們可以這樣做。假設我們有一個讀取行的 goroutine，一個在 XML 元素中處理它們，一個對元素執行某些操作。
// A buffered channel for input
lines := make(chan []byte, 10)
elements := make(chan *RS)
var wg sync.WaitGroup
// Worker goroutine, turn lines into RS structs
wg.Add(1)
go func() {
? ? defer wg.Done()
? ? defer close(elements)
? ? for line := range lines {
? ? ? ? if line[2] == 32 {
? ? ? ? ? ? fmt.Println("Begin")
? ? ? ? ? ? fmt.Println(string(line))
? ? ? ? ? ? fmt.Println("End")
? ? ? ? ? ? rs := &RS{}
? ? ? ? ? ? xml.Unmarshal(line, &rs)
? ? ? ? ? ? elements <- rs
? ? ? ? }
? ? }
}()
// File reader
go func() {
? ? defer close(lines)
? ? for scanner.Scan() {
? ? ? ? lines <- scanner.Bytes()
? ? }
? ? if err := scanner.Err(); err != nil {
? ? ? ? log.Fatal(err)
? ? }
}()
// Element reader
wg.Add(1)
go func() {
? ? defer wg.Done()
? ? for element := range elements {
? ? ? ? fmt.Println(element)
? ? }
}()
wg.Wait()
這會產生空結構，因為您正試圖將單獨的 XML 行放入表示完整標記的結構中<relationship>。但它演示了如何向鏈中添加更多工人。
蕪湖不蕪
TA貢獻1796條經驗 獲得超7個贊
我不知道這是否對您有幫助，但您的條件if p2Rc[2]==114永遠不會滿足，然后您繼續并開始收聽頻道。從不接收輸入。
我認為這里的主要問題是這個（代碼）是做什么的？上述條件屬于該過程的何處？如果這很清楚，我可以更新一個更好的回應。
- 2 回答
- 0 關注
- 134 瀏覽
添加回答
舉報
0/150
提交
取消