1 前置条件
Golang基本情况自行baidu/google
1.1 环境与工具
IDE：liteide (windows )、mingw-w64 (gcc)
DB：SQL Server 2008 R2
MQ: Kakfa
1.2 环境变量
Golang 需要将GOPATH添加到环境变量中
1.3 版本查看
Cmd中输入 go version 可以查看当前golang版本
2 示例程序功能简介
2.1 主功能
- 查询数据表A获取记录的消息编号MSG-Id.
- 从数据表B查询大于MSG-Id的100条连续数据.
- 将查询到的100条数据依次序列化成JSON数据，推送到消息队列Kafka中（供其它软件使用）.
- 100条推送成功后将最大的MSG-Id更新到数据表A.
2.2 辅助项
- 循环执行，每执行主功能一次，挂起2秒
- 记录关键日志到文件
- 从配置文件读取信息
- 在Console上显示实时信息
3 代码描述
3.1 目录结构与程序包
MainExe：主项目模块的目录
Domain领域层目录
xxDomain.go 领域层源码文件（Package yyDomain）
Model模型层目录
xxModel.go 模型层源码文件（Package yyModel）
Main.go:主函数源码文件，Package Main
Common：通用/公共功能模块目录
MQ_kafka帮助类子目录
Sarama.go: kafka帮助类源码（package KafkaHelper）
DB_SQLServer帮助类子目录
SQLServer.go: SQLServer帮助类源码（package SQLHelper）
File_CFG帮助类子目录
CFG.go: 配置文件操作帮助类源码（package CFGHelper）
在Goalng中源码的物理文件是按照目录（Directory）来管理, 每个*.go文件都存放在一个目录下，一个目录下的所有源文件只能属于相同的包package。在逻辑上，源码使用包（package）这种语法元素来组织源码，所有语法可见性均定义在package这个级别，每个*.go文件都属于一个Package。
导入：Import（”目录/子目录”）
程序调用: PackageName.xxxMethod/yyyFunction
3.2 源码文件SQLServer.go
这里用到了go-mssqldb这个包，在命令行cmd使用go get github.com/denisenkom/go-mssqldb 可以将程序包下载到本地
这个包在import时，前面加了横线，说明即使用【import _ 包路径】只是引用该包，仅仅是为了调用init()函数，所以无法通过包名来调用包中的其他函数。
package SQLHelper import ( "Common/File_CFG" "database/sql" "encoding/json" "fmt" "log" "time" _ "github.com/denisenkom/go-mssqldb" ) func QueryDataReturnString(strSQL string /*, connInstance *sql.DB*/) (strValue string, err error) { ip, port, user, password, database, err := Base_ConfigFileHelper.GetSQLServerConfig() if err != nil { log.Fatal("Read Config File failed:", err.Error()) } connString := fmt.Sprintf("server=%s;port%s;database=%s;user id=%s;password=%s", ip, port, database, user, password) connInstance, err := sql.Open("mssql", connString) if err != nil { log.Fatal("Open Connection failed:", err.Error()) } defer connInstance.Close() var strLastId string errTemp := connInstance.QueryRow(strSQL).Scan(&strLastId) if errTemp != nil { log.Fatal("Query failed:", errTemp.Error()) } return strLastId, errTemp } func QueryDataReturnArray(strSQL string /*, connInstance *sql.DB*/) (dataRows *sql.Rows) { ip, port, user, password, database, err := Base_ConfigFileHelper.GetSQLServerConfig() if err != nil { log.Fatal("Read Config File failed:", err.Error()) } connString := fmt.Sprintf("server=%s;port%d;database=%s;user id=%s;password=%s", ip, port, database, user, password) // if isdebug { // fmt.Println(connString) // } connInstance, err := sql.Open("mssql", connString) if err != nil { log.Fatal("Open Connection failed:", err.Error()) } defer connInstance.Close() stmt, err := connInstance.Prepare(strSQL) if err != nil { log.Fatal("Prepare failed:", err.Error()) } rowsArray, err := stmt.Query() if err != nil { log.Fatal("Query failed:", err.Error()) } stmt.Close() //defer rowsArray.Close() return rowsArray } func ExeNoQuery(strSQL string /*, connInstance *sql.DB*/) (returnResult int64, err error) { ip, port, user, password, database, err := Base_ConfigFileHelper.GetSQLServerConfig() if err != nil { log.Fatal("Read Config File failed:", err.Error()) } connString := fmt.Sprintf("server=%s;port%d;database=%s;user id=%s;password=%s", ip, port, database, user, password) connInstance, err := sql.Open("mssql", connString) if err != nil { log.Fatal("Open Connection failed:", err.Error()) } defer connInstance.Close() stmt, err := connInstance.Prepare(strSQL) if err != nil { log.Fatal("Prepare failed:", err.Error()) } defer stmt.Close() returnRows, err := stmt.Exec() if err != nil { log.Fatal("Query failed:", err.Error()) } //我们可以获得插入的id //id, err := rs.LastInsertId() //可以获得影响行数 result, err := returnRows.RowsAffected() return result, err }
// 参考资料
//Go-SQLServer https://www.cnblogs.com/dfsxh/p/10207866.html
//Go-SQLServer https://www.cnblogs.com/Liang2790912648/p/10604310.html
//Go-SQLServer https://www.jianshu.com/p/136ac470fc17
//Go-SQLServer https://blog.csdn.net/chen_peng7/article/details/89317922
//GO-MySQL(scan) https://zhidao.baidu.com/question/1759806412743140028.html
3.3 源码文件Sarama.go
package KafkaHelper import ( "fmt" "time" "github.com/Shopify/sarama" ) func CreateInstance() (syncClient sarama.SyncProducer, err error) { //初始化配置 newConfig := sarama.NewConfig() newConfig.Producer.RequiredAcks = sarama.WaitForAll newConfig.Producer.Partitioner = sarama.NewRandomPartitioner newConfig.Producer.Return.Successes = true //生产者 client, err := sarama.NewSyncProducer([]string{"172.16.77.53:9092"}, newConfig) if err != nil { fmt.Println("producer close,err:", err) return nil, err } else { return client, err } } func SendMessage(syncClient sarama.SyncProducer, InMessage string) { msg := &sarama.ProducerMessage{} msg.Topic = "02.dbo " msg.Value = sarama.StringEncoder(InMessage) //发送消息 pid, offset, err := syncClient.SendMessage(msg) if err != nil { fmt.Println(time.Now().Format("2006-01-02 15:04:05")+" send failed: ", err) } else { fmt.Printf(time.Now().Format("2006-01-02 15:04:05")+" send success:1 pid:%v offset:%v\n", pid, offset) } }
因为在.net core 环境下用Confluent成功，本想goalng用Confluent package继续测试，但是下载这个包很困难（墙），所以放弃了。
// 参考资料
//Go-Kafka https://www.cnblogs.com/vincenshen/p/9824486.html
//Go-Kafka https://www.cnblogs.com/pyyu/p/8371649.html
//Go-Kafka https://blog.csdn.net/lanyang123456/article/details/78377152
//Go-Kafka https://www.jianshu.com/p/1136f37cc419
3.4 源码文件CFG.go
package CFGHelper import ( "fmt" "github.com/Unknwon/goconfig" ) func GetSQLServerConfig() (ip string, port string, username string, password string, dbname string, err error) { cfgFile, err := goconfig.LoadConfigFile("conf.ini") if err != nil { fmt.Println(err) } ipGet, err := cfgFile.GetValue("DB_SQLServer", "ip") if err != nil { fmt.Println(err) } portGet, err := cfgFile.GetValue("DB_SQLServer", "port") if err != nil { fmt.Println(err) } usernameGet, err := cfgFile.GetValue("DB_SQLServer", "username") if err != nil { fmt.Println(err) } passwordGet, err := cfgFile.GetValue("DB_SQLServer", "password") if err != nil { fmt.Println(err) } dbnameGet, err := cfgFile.GetValue("DB_SQLServer", "dbname") if err != nil { fmt.Println(err) } return ipGet, portGet, usernameGet, passwordGet, dbnameGet, err } func GetKafkaConfig() (ip string, topic string, groupid string, err error) { cfgFile, err := goconfig.LoadConfigFile("../conf.ini") if err != nil { fmt.Println(err) } ipGet, err := cfgFile.GetValue("MQ_Kafka", "ip") if err != nil { fmt.Println(err) } topicGet, err := cfgFile.GetValue("MQ_Kafka", "topic") if err != nil { fmt.Println(err) } groupidGet, err := cfgFile.GetValue("MQ_Kafka", "groupid") if err != nil { fmt.Println(err) } return ipGet, topicGet, groupidGet, err }
3.4 源码文件main.go
import 里面有很多测试导入的包，如果报错就屏蔽掉
package main import ( "MainMenthod/Domain" "fmt" "io" "net/http" "os" "reflect" "time" ) func main() { mainDomain.DoWork_Init() for true { //var currentMonth：=time.Now().String() mainDomain.DoWork_SendMSG() time.Sleep(2 * time.Second) } mainDomain.DoWork_Clean() //testReflect() //testgoroutine() //testhttpget() fmt.Println("program exit") }
3.4 源码文件xxModel.go
package yyModel type BrushCardModel struct { Odid int32 StepCode string EmpCode string SysCode string Sort string OrderCode string LoadDate string WorkHours int32 }
3.4 源码文件yyDomain.go
package mainDomain import ( "CommonCode/DB_SQLServer" "CommonCode/MQ_Kafka" "MainMenthod/Model" "database/sql" "encoding/json" "fmt" "log" "os" "strconv" "strings" "time" "github.com/Shopify/sarama" _ "github.com/denisenkom/go-mssqldb" ) var kafkaProducer sarama.SyncProducer func DoWork_Init() { producer, err := KafkaHelper.CreateInstance() if err != nil { log.Fatal("Create Kafka Producer failed:", err.Error()) fmt.Println("Create Kafka Producer failed" + err.Error()) return } else { kafkaProducer = producer } } func DoWork_Clean() { kafkaProducer.Close() } func DoWork_SendMSG() { //固定的时间；“2006-01-02” file := "./" + time.Now().Format("20060102") + ".txt" logFile, err := os.OpenFile(file, os.O_RDWR|os.O_CREATE|os.O_APPEND, os.ModePerm) if nil != err { panic(err) } defer logFile.Close() loger := log.New(logFile, "", log.Ldate|log.Ltime|log.Lshortfile) var strSQL_LastId string = "select LastMSGId from A where MSGTable=\'B\'" LastMSGIdOfBC, err := SQLHelper.QueryDataReturnString(strSQL_LastId /*, connInstance*/) if err != nil { fmt.Println("Query LastMSGId failed" + err.Error()) loger.Output(2, "Query LastMSGId failed"+err.Error()+"\r\n") } var builderSQL strings.Builder builderSQL.WriteString(" select top 100 MaterialCardBrushID,Odid,SysCode,OrderCode,") builderSQL.WriteString(" EmpCode,StepCode,Sort,WorkHours, ") builderSQL.WriteString(" CONVERT(varchar(100), LoadDate, 20) ") builderSQL.WriteString(" from B where MaterialCardBrushID >" + LastMSGIdOfBC) builderSQL.WriteString(" order by MaterialCardBrushID") //fmt.Println(builderSQL.String()) var rowsArray *sql.Rows rowsArray = SQLHelper.QueryDataReturnArray(builderSQL.String() /*, connInstance*/) var materialCardBrushID int for rowsArray.Next() { ins_BrushCard := &mainModel.BrushCardModel{} err = rowsArray.Scan(&materialCardBrushID, &ins_BrushCard.Odid, &ins_BrushCard.SysCode, &ins_BrushCard.OrderCode, &ins_BrushCard.EmpCode, &ins_BrushCard.StepCode, &ins_BrushCard.Sort, &ins_BrushCard.WorkHours, &ins_BrushCard.LoadDate) if err != nil { fmt.Println("Get Data from Row failed" + err.Error()) loger.Output(2, "Get Data from Row failed"+err.Error()+"\r\n") break } else { jsonOfBrushCard, err := json.Marshal(ins_BrushCard) if err != nil { fmt.Println("生成json字符串错误" + err.Error()) loger.Output(2, "生成json字符串错误"+err.Error()+"\r\n") } var strValue string = string(jsonOfBrushCard) loger.Output(2, strValue+"\r\n") KafkaHelper.SendMessage(kafkaProducer, strValue) } } rowsArray.Close() var strSQL_Update string = "update SendLastMSGFlag set LastMSGId=" + strconv.Itoa(materialCardBrushID) + " where MSGTable=\'B\'" result, err := SQLHelper.ExeNoQuery(strSQL_Update /*, connInstance*/) if err != nil { //log.Fatal("Update failed:materialCardBrushID="+strconv.Itoa(materialCardBrushID), err.Error()) fmt.Println(time.Now().Format("2006-01-02 15:04:05") + " LastMSGId:" + strconv.Itoa(materialCardBrushID) + " failed:" + err.Error()) loger.Output(2, " LastMSGId:"+strconv.Itoa(materialCardBrushID)+" failed:"+err.Error()+"\r\n") } else { fmt.Println(time.Now().Format("2006-01-02 15:04:05") + " update success:" + strconv.FormatInt(result, 10) + " LastMSGId:" + strconv.Itoa(materialCardBrushID)) loger.Output(2, " update success:"+strconv.FormatInt(result, 10)+" LastMSGId:"+strconv.Itoa(materialCardBrushID)+"\r\n") } }
// 参考资料
// Go-JSON https://www.cnblogs.com/pyyu/p/8309917.html
// Go-JSON https://www.jianshu.com/p/3534532e06ed
// Go-JSON https://studygolang.com/articles/8526 SQL结果集-JSON
// Go-base https://www.jianshu.com/p/dbb39c69e3f1 类型转换string、int、int64
// Go-base https://www.cnblogs.com/smallleiit/p/10694481.html 类型转换string、int、int64
// Go-base https://www.cnblogs.com/pyyu/p/8258133.html 自定义类型struct
// Go-base https://studygolang.com/articles/3383 log简介
3.4 配置文件conf.ini
[DB_SQLServer] ip=127.0.0.1 port=1433 username=sa password=123456 dbname=test [MQ_Kafka] ip=127.0.0.1:9092 topic=topcoco groupid=zzc
对于熟悉一门编程语言，学习另外一种语言，总有一些困难：
- 首先是开始时的开发--编译-运行环境，也就是hello world，以及调试：需要搜集大量资料或教程才能敲门
- 其次是习惯，从面向过程转面向对象或者反之，都需要克服习惯
- 最后，是做一个工作上用到的，不大的有用的程序，除非你的兴趣特别强
