用golang遍历证券软件通达信日线目录,把所有日线数据发送到mongodb数据库中

package main
import (
 "bytes"
 "encoding/binary"
 "fmt"
 "gopkg.in/mgo.v2"
 //"gopkg.in/mgo.v2/bson"
 //"log"
 "io"
 "net"
 "os"
 "time"
)
const (
 BZ = "PUTDATA"
 MongodbUrl = "localhost:27017"
)
type StockFile struct {
 Date, Open, High, Low, Close int32
 Amount float32
 Vol int32
 Preclose int32
}
type StockData struct {
 Code string
 Date int32
 Open, High, Low, Close, Amount float32
 Vol, Preclose int32
}
func main() {
 //在7070端口监听
 tcpAddr, err := net.ResolveTCPAddr("tcp", ":7076")
 checkError(err)
 listener, err1 := net.ListenTCP("tcp", tcpAddr)
 checkError(err1)
 fmt.Println("服务器准备就绪")
 session, err := mgo.Dial(MongodbUrl)
 if err != nil {
 panic(err)
 }
 // Optional. Switch the session to a monotonic behavior.
 session.SetMode(mgo.Monotonic, true)
 defer session.Close()
 //c := session.DB("stock").C("day")
 for {
 //等待客户端的连接
 conn, err2 := listener.Accept()
 if err != nil {
 /*通常服务端为一个服务,不会因为错误而退出。出错后,继续等待下一个连接请求*/
 fmt.Println(err2)
 continue
 }
 fmt.Println("收到客户端的请求")
 go ServeClient(conn, session)
 }
}
func ServeClient(conn net.Conn, session *mgo.Session) {
 defer conn.Close()
 defer fmt.Println("接收数据完成")
 str := ReadData(conn)
 if str == BZ {
 SendData(conn, "OK")
 fmt.Println("开始接受数据...")
 Code := ReadData(conn)
 c := session.DB("stock").C(Code)
 SendData(conn, "CODEOK")
 st := time.Now().UnixNano()
 //显示掺入前集合中元素数目
 countNum, err := c.Count()
 if err != nil {
 panic(err)
 }
 fmt.Println("mongodb连接成功,这个数据库的对象总计: ", countNum)
 ReadHq(conn, c, Code)
 fmt.Printf("执行程序花费的时间为(毫秒): %d \r\n", (time.Now().UnixNano()-st)/int64(time.Millisecond))
 } else {
 SendData(conn, "bad connetct")
 }
 return
}
/*读取验证数据*/
func ReadData(conn net.Conn) string {
 var data bytes.Buffer
 var buf [512]byte
 for {
 n, err := conn.Read(buf[0:])
 if err != nil {
 fmt.Println(err)
 return ""
 }
 //我们的数据以0做为结束的标记
 if buf[n-1] == 0 {
 //n-1去掉结束标记0
 data.Write(buf[0 : n-1])
 break
 } else {
 data.Write(buf[0:n])
 }
 }
 return string(data.Bytes())
}
func ReadHq(conn net.Conn, c *mgo.Collection, Code string) {
 var buf [32]byte
 for {
 _, err := conn.Read(buf[0:32])
 if err == io.EOF {
 fmt.Println("此个文件传输结束")
 break
 }
 if err != nil {
 fmt.Println(err)
 return
 }
 b_buf := bytes.NewBuffer(buf[0:32])
 var x StockFile
 var y StockData
 binary.Read(b_buf, binary.LittleEndian, &x) //binary.LittleEndian 是内存中的字节序的概念,就是把低字节的放到了后面。网络传输一般用BigEndian,内存字节序和cpu有关,编程时要转化。
 y.Date = x.Date
 y.Open = float32(x.Open) / 100
 y.High = float32(x.High) / 100
 y.Low = float32(x.Low) / 100
 y.Close = float32(x.Close) / 100
 y.Preclose = x.Preclose
 y.Amount = x.Amount
 y.Vol = x.Vol
 //y.Code = Code
 fmt.Println(y)
 err = c.Insert(&y)
 if err != nil {
 panic(err)
 }
 }
 return
}
func SendData(conn net.Conn, data string) {
 buf := []byte(data)
 /*向 byte 字节里添加结束标记*/
 buf = append(buf, 0)
 _, err := conn.Write(buf)
 if err != nil {
 fmt.Println(err)
 }
}
func checkError(err error) {
 if err != nil {
 fmt.Println(err)
 os.Exit(0)
 }
}
client
package main
import (
 "bytes"
 "fmt"
 "io"
 //"io/ioutil"
 "net"
 "os"
 "strings"
 "time"
)
const (
 BZ = "PUTDATA"
)
func main() {
 tcpAddr, err := net.ResolveTCPAddr("tcp", "192.168.1.6:7076")
 checkError(err)
 //遍历目录
 dir, err := os.OpenFile("/Users/wangyan/go/src/hangqing/day", os.O_RDONLY, 0666)
 if err != nil {
 fmt.Println(err.Error())
 return
 }
 arrFile, err1 := dir.Readdir(0)
 if err1 != nil {
 fmt.Println(err1.Error())
 return
 }
 for _, v := range arrFile {
 //tcp 握手
 conn, err := net.DialTCP("tcp", nil, tcpAddr)
 checkError(err)
 defer conn.Close()
 SendData(conn, BZ)
 s := ReadData(conn)
 if s == "OK" {
 fmt.Println("连接成功")
 Code := strings.TrimRight(v.Name(), ".day")
 SendData(conn, Code)
 if ReadData(conn) == "CODEOK" {
 fmt.Println("传输证券代码完成")
 f, err := os.OpenFile("/Users/wangyan/go/src/hangqing/day/"+v.Name(), os.O_RDONLY, 0666)
 checkError(err)
 defer f.Close()
 //每次传递32字节
 buf := make([]byte, 32)
 //重置文件指针,否则读不到内容的。
 f.Seek(0, os.SEEK_SET)
 for {
 n, ferr := f.Read(buf)
 if ferr != nil && ferr != io.EOF {
 fmt.Println(ferr.Error())
 break
 }
 if ferr == io.EOF {
 fmt.Println(ferr.Error())
 break
 }
 if n == 0 {
 break
 }
 //str += string(buf[0:n])
 SendHq(conn, buf)
 }
 /*一次读完整个文件
 buf, err := ioutil.ReadAll(f)
 checkError(err)
 SendHq(conn, buf)
 */
 fmt.Println("传输数据完成:", v.Name())
 conn.Close()
 f.Close()
 } else {
 fmt.Println("code not translat")
 conn.Close()
 }
 } else {
 fmt.Println("连接失败")
 conn.Close()
 }
 time.Sleep(1 * time.Second)
 }
}
/*读取数据*/
func ReadData(conn net.Conn) string {
 var data bytes.Buffer
 var buf [512]byte
 for {
 n, err := conn.Read(buf[0:])
 if err != nil {
 fmt.Println(err)
 return ""
 }
 //我们的数据以0做为结束的标记
 if buf[n-1] == 0 {
 //n-1去掉结束标记0
 data.Write(buf[0 : n-1])
 break
 } else {
 data.Write(buf[0:n])
 }
 }
 return string(data.Bytes())
}
func SendData(conn net.Conn, data string) {
 buf := []byte(data)
 /*向 byte 字节里添加结束标记*/
 buf = append(buf, 0)
 _, err := conn.Write(buf)
 if err != nil {
 fmt.Println(err)
 }
}
func SendHq(conn net.Conn, data []byte) {
 _, err := conn.Write(data)
 if err != nil {
 fmt.Println(err)
 }
}
func checkError(err error) {
 if err != nil {
 fmt.Println(err)
 os.Exit(0)
 }
}

有疑问加站长微信联系(非本文作者)

• 请尽量让自己的回复能够对别人有帮助
• 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
• 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
• 图片支持拖拽、截图粘贴等方式上传